From c22e27d0399a3cf00552f220c7f75e79fea3c748 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:06:26 -0400 Subject: [PATCH 01/11] feat(engine): a registered descriptor can be given its constructor later The declaration-registers half the wheel needs: a descriptor registered without a constructor gains one through a single entry, refusing a path that already has one with the two-classes-one-path text and an unknown path by name. Co-Authored-By: Claude Opus 5 --- .github/workflows/test.yml | 3 + .../processors/processor_instance_factory.rs | 140 ++++++++++++++++++ xtask/src/main.rs | 3 + 3 files changed, 146 insertions(+) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index dfcf71da0..4029f09d9 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -290,6 +290,9 @@ jobs: core::json_schema::port_rendering_tests::port_descriptor_output_carries_no_type_key \ core::json_schema::config_schema_rendering_tests::a_registered_descriptors_config_schema_reaches_the_rendering_unchanged \ core::json_schema::config_schema_rendering_tests::a_descriptor_carrying_no_config_schema_renders_no_key_rather_than_a_null \ + core::processors::processor_instance_factory::tests::a_constructor_installs_onto_a_descriptor_registered_without_one \ + core::processors::processor_instance_factory::tests::installing_onto_a_path_that_already_has_a_constructor_is_refused \ + core::processors::processor_instance_factory::tests::installing_onto_an_unregistered_path_is_refused_naming_the_path \ core::json_schema::port_rendering_tests::a_contract_bearing_port_renders_its_contract_beside_the_four \ core::json_schema::port_rendering_tests::a_port_declaring_the_sentinel_renders_it_as_a_whole_contract \ core::json_schema::port_rendering_tests::a_declared_contract_survives_the_descriptor_to_port_info_hop \ diff --git a/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs b/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs index 6b4c0ad11..cdaeac58c 100644 --- a/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs +++ b/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs @@ -436,6 +436,48 @@ impl ProcessorInstanceFactory { Ok(()) } + /// Give a descriptor that registered without a constructor the one that + /// can build it — what a Python class's first add supplies, after its + /// decorator registered the descriptor at import. + /// + /// Succeeds only on a path holding a descriptor and no constructor. A path + /// that already has one is two classes claiming one import path, refused + /// with the same text a second registration meets; a path nobody + /// registered is refused by name rather than registered here, because the + /// descriptor is the decorator's to write. + pub fn install_constructor_for_registered_descriptor( + &self, + processor_class_import_path: &ProcessorClassImportPath, + constructor: DynamicProcessorConstructorFn, + ) -> Result<()> { + // The same outer-to-inner order a registration takes, so an install + // racing one cannot interleave between their check and their claim. + let descriptors = self.descriptors.read(); + if !descriptors.contains_key(processor_class_import_path) { + return Err(Error::ProcessorNotFound(format!( + "no descriptor is registered for processor type '{processor_class_import_path}', so there is nothing to install a constructor onto. A Python class registers its descriptor when its `@processor` decorator runs, so a path missing here names a class this process never imported." + ))); + } + + let mut registrations = self.registrations.write(); + if registrations.contains_key(processor_class_import_path) { + return Err(duplicate_class_import_path(processor_class_import_path)); + } + registrations.insert( + processor_class_import_path.clone(), + RegistrationKind::LegacyDyn { constructor }, + ); + drop(registrations); + drop(descriptors); + + tracing::info!( + processor_class_import_path = processor_class_import_path.as_str(), + "[install_constructor_for_registered_descriptor] a registered descriptor gained its constructor" + ); + + Ok(()) + } + /// Install the resolver consulted when an add names a type nobody registered. pub fn set_unregistered_processor_type_resolver( &self, @@ -795,6 +837,104 @@ mod tests { } } + /// The declaration-registers path: a descriptor lands at import with no + /// constructor, and the first add installs one onto it rather than + /// registering a second time. + #[test] + fn a_constructor_installs_onto_a_descriptor_registered_without_one() { + let factory = ProcessorInstanceFactory::new(); + let path = "my_app.filters:BlurProcessor"; + + factory + .register_descriptor_only(descriptor_for(path)) + .expect("the decorator's descriptor-only registration succeeds"); + assert!( + !factory.can_create(&class_import_path(path)), + "a descriptor-only registration has nothing to construct with" + ); + + factory + .install_constructor_for_registered_descriptor( + &class_import_path(path), + Box::new(|_node| Err(Error::Configuration("unreachable".into()))), + ) + .expect("the first add installs its constructor"); + + assert!(factory.can_create(&class_import_path(path))); + assert_eq!( + factory.list_registered().len(), + 1, + "installing a constructor adds no second catalog entry" + ); + assert!( + factory.port_info(&class_import_path(path)).is_some(), + "the descriptor the decorator registered keeps its port info" + ); + } + + /// Two classes claiming one import path are the same collision whichever + /// half of the registration arrives second, so the install refuses with the + /// text the registration would have. + #[test] + fn installing_onto_a_path_that_already_has_a_constructor_is_refused() { + let factory = ProcessorInstanceFactory::new(); + let path = "my_app.filters:BlurProcessor"; + + factory + .register_dynamic( + descriptor_for(path), + Box::new(|_node| Err(Error::Configuration("the first".into()))), + ) + .expect("the first registration succeeds"); + + let refusal = factory + .install_constructor_for_registered_descriptor( + &class_import_path(path), + Box::new(|_node| Err(Error::Configuration("the second".into()))), + ) + .expect_err("a path that already has a constructor must be refused"); + + let Error::Configuration(message) = refusal else { + panic!("expected Configuration; got {refusal:?}"); + }; + assert!( + message.contains(path), + "the refusal must name the contested path; got: {message}" + ); + assert!( + message.contains("importlib.reload"), + "the refusal must carry the two-classes-one-path text; got: {message}" + ); + } + + /// Nothing registered the descriptor, so there is nothing to install onto: + /// silently registering here would let a class the decorator never saw + /// reach the catalog by a different door. + #[test] + fn installing_onto_an_unregistered_path_is_refused_naming_the_path() { + let factory = ProcessorInstanceFactory::new(); + let path = "my_app.filters:NeverDeclared"; + + let refusal = factory + .install_constructor_for_registered_descriptor( + &class_import_path(path), + Box::new(|_node| Err(Error::Configuration("unreachable".into()))), + ) + .expect_err("an unknown path must be refused"); + + let Error::ProcessorNotFound(message) = refusal else { + panic!("expected ProcessorNotFound; got {refusal:?}"); + }; + assert!( + message.contains(path), + "the refusal must name the path nobody registered; got: {message}" + ); + assert!( + factory.descriptor(&class_import_path(path)).is_none(), + "a refused install must leave the registry untouched" + ); + } + #[test] fn an_unregistered_path_is_absent_from_every_map() { let factory = ProcessorInstanceFactory::new(); diff --git a/xtask/src/main.rs b/xtask/src/main.rs index d44842814..94c9a6915 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -389,6 +389,9 @@ fn run_local_ci_gates(workspace_root: &Path) -> Result<()> { "core::json_schema::port_rendering_tests::port_descriptor_output_carries_no_type_key", "core::json_schema::config_schema_rendering_tests::a_registered_descriptors_config_schema_reaches_the_rendering_unchanged", "core::json_schema::config_schema_rendering_tests::a_descriptor_carrying_no_config_schema_renders_no_key_rather_than_a_null", + "core::processors::processor_instance_factory::tests::a_constructor_installs_onto_a_descriptor_registered_without_one", + "core::processors::processor_instance_factory::tests::installing_onto_a_path_that_already_has_a_constructor_is_refused", + "core::processors::processor_instance_factory::tests::installing_onto_an_unregistered_path_is_refused_naming_the_path", "core::json_schema::port_rendering_tests::a_contract_bearing_port_renders_its_contract_beside_the_four", "core::json_schema::port_rendering_tests::a_port_declaring_the_sentinel_renders_it_as_a_whole_contract", "core::json_schema::port_rendering_tests::a_declared_contract_survives_the_descriptor_to_port_info_hop", From 7fd6a06081267f07a5f9c06e6692375d5b5c98b5 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:09:44 -0400 Subject: [PATCH 02/11] feat(wheel)!: @processor registers its descriptor when it runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Decorating a class puts it in the processor catalog, so an agent reads an effect an app imported and never added. The constructor still arrives at the first add, installed onto that descriptor. A decoration inside a helper process registers nothing, and so does a class no interpreter could import — `rt.add` is where that is refused, with the fix named. `description` falls back to the class's docstring. Co-Authored-By: Claude Opus 5 --- .../python/streamlib/_engine.pyi | 17 ++ .../streamlib/_processor_declaration.py | 15 +- sdk/streamlib-python-wheel/src/lib.rs | 8 + .../src/python_helper_process_spawn_host.rs | 10 +- .../src/python_processor_registration.rs | 59 +++++- .../tests/test_declaration_registers.py | 169 ++++++++++++++++++ 6 files changed, 269 insertions(+), 9 deletions(-) create mode 100644 sdk/streamlib-python-wheel/tests/test_declaration_registers.py diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index 8745b3abc..e1aa228ac 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi +++ b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi @@ -1640,6 +1640,23 @@ def capability_extension_host_for_the_helper_process( ) -> CapabilityExtensionHost: """Mint the host `distribution`'s hook is handed in a helper process.""" +def register_declared_processor_class(processor_class: type) -> None: + """Register the descriptor `@processor` has just stamped onto a class. + + Called by the decorator and nowhere else, so the class is in the processor + catalog from the moment its module is imported; the constructor arrives at + the first `Runtime.add`. A class decorated inside a helper process + registers nothing — a helper hosts no graph — and so does one no + interpreter could import, which `Runtime.add` refuses by name. + """ + +def processor_class_import_paths_registered_in_this_process() -> list[str]: + """Every processor class import path the calling process has registered. + + The catalog `GET /api/registry` renders, readable in a process that serves + no control plane. + """ + def monotonic_now_ns() -> int: """Current monotonic time in nanoseconds via `clock_gettime(CLOCK_MONOTONIC)`.""" diff --git a/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py b/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py index d6c092ea9..9d4ae62c4 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py +++ b/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py @@ -22,6 +22,7 @@ import typing from typing import Any, Callable, Optional, TypeVar +from ._engine import register_declared_processor_class from ._processor_config_schema import ( derive_config_class_json_schema, json_schema_for_a_processor_declaring_no_config, @@ -355,6 +356,12 @@ def processor( Schema is derived from its annotations and defaults and published in the processor catalog, which is how an agent learns the keys before adding the node. + + Decorating registers the class's descriptor — identity, description, ports + and config schema — so the class is in that catalog from the moment its + module is imported, whether or not anything ever adds it. The constructor + arrives at the first `rt.add`. `description` falls back to the class's + docstring when it is not given. """ if isinstance(processor_class, type): return _declare_processor( @@ -405,7 +412,9 @@ def _declare_processor( if config_class is None else derive_config_class_json_schema(config_class) ) - processor_class.__streamlib_processor_description__ = description # type: ignore[attr-defined] + processor_class.__streamlib_processor_description__ = ( # type: ignore[attr-defined] + description or inspect.getdoc(processor_class) or "" + ) processor_class.__streamlib_processor_execution__ = _resolve_execution( # type: ignore[attr-defined] execution, interval_ms, processor_class, has_input_ports=bool(input_ports) ) @@ -414,6 +423,10 @@ def _declare_processor( ) processor_class.__streamlib_processor_input_ports__ = input_ports # type: ignore[attr-defined] processor_class.__streamlib_processor_output_ports__ = output_ports # type: ignore[attr-defined] + + # After the attributes and not before: the native half reads exactly them + # to build the descriptor it registers. + register_declared_processor_class(processor_class) return processor_class diff --git a/sdk/streamlib-python-wheel/src/lib.rs b/sdk/streamlib-python-wheel/src/lib.rs index b1eb6391d..c3e66bb89 100644 --- a/sdk/streamlib-python-wheel/src/lib.rs +++ b/sdk/streamlib-python-wheel/src/lib.rs @@ -100,6 +100,14 @@ fn _engine(module: &Bound<'_, PyModule>) -> PyResult<()> { python_capability_extension_host::capability_extension_host_for_the_helper_process, module )?)?; + module.add_function(wrap_pyfunction!( + python_processor_registration::register_declared_processor_class, + module + )?)?; + module.add_function(wrap_pyfunction!( + python_processor_registration::processor_class_import_paths_registered_in_this_process, + module + )?)?; module.add_function(wrap_pyfunction!(python_logging::monotonic_now_ns, module)?)?; module.add_function(wrap_pyfunction!(python_logging::log_event, module)?)?; module.add_function(wrap_pyfunction!( diff --git a/sdk/streamlib-python-wheel/src/python_helper_process_spawn_host.rs b/sdk/streamlib-python-wheel/src/python_helper_process_spawn_host.rs index 216f04710..2add15ef8 100644 --- a/sdk/streamlib-python-wheel/src/python_helper_process_spawn_host.rs +++ b/sdk/streamlib-python-wheel/src/python_helper_process_spawn_host.rs @@ -34,6 +34,11 @@ use streamlib::sdk::processors::{DynGeneratedProcessor, OutOfProcessLinkWiringEn /// The module CPython is launched with in a helper process. const HELPER_PROCESS_MODULE: &str = "streamlib._helper"; +/// The environment variable carrying the class import path a helper process +/// hosts — set here and nowhere else, so its presence is what tells code +/// running inside a child that it is one. +pub(crate) const HELPER_PROCESS_ENTRYPOINT_ENVIRONMENT_VARIABLE: &str = "STREAMLIB_ENTRYPOINT"; + /// How long the child has to import the user's class, open its ports, run /// `setup` and report ready before this host gives up and kills it. /// @@ -178,7 +183,10 @@ impl PythonHelperProcessSpawnHostProcessor { // only send the child looking for the wrong standard library. .env_remove("PYTHONHOME") .env("PYTHONPATH", self.child_python_path()) - .env("STREAMLIB_ENTRYPOINT", &self.processor_class_import_path) + .env( + HELPER_PROCESS_ENTRYPOINT_ENVIRONMENT_VARIABLE, + &self.processor_class_import_path, + ) .env("STREAMLIB_PROCESSOR_ID", &self.processor_id) .env("STREAMLIB_RUNTIME_ID", runtime_id) .env( diff --git a/sdk/streamlib-python-wheel/src/python_processor_registration.rs b/sdk/streamlib-python-wheel/src/python_processor_registration.rs index d8f62a919..43462874f 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_registration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_registration.rs @@ -3,9 +3,12 @@ //! Making a Python class a processor type the engine can instantiate. //! +//! Registration arrives in two halves. `@processor` registers the descriptor +//! when it runs, so the class is in the catalog an agent reads before anything +//! adds it; the first add installs the constructor onto that descriptor. //! Registration is per process and idempotent per identity: `rt.add(Blur)` -//! called twice registers `Blur` once and adds two processors to the graph, -//! each with its own configuration and its own instance of the class. +//! called twice installs `Blur`'s constructor once and adds two processors to +//! the graph, each with its own configuration and its own instance of the class. use std::collections::HashMap; use std::sync::{Mutex, OnceLock}; @@ -15,8 +18,11 @@ use pyo3::prelude::*; use streamlib::sdk::descriptors::ProcessorClassImportPath; use streamlib::sdk::processors::PROCESSOR_REGISTRY; -use crate::python_helper_process_spawn_host::spawn_host_for_processor_node; +use crate::python_helper_process_spawn_host::{ + HELPER_PROCESS_ENTRYPOINT_ENVIRONMENT_VARIABLE, spawn_host_for_processor_node, +}; use crate::python_processor_declaration::PythonProcessorDeclaration; +use crate::python_processor_import_path::processor_class_import_path; /// Which Python class each registered import path was registered from. /// @@ -72,7 +78,6 @@ pub(crate) fn register_processor_class( }; } - let descriptor = declaration.descriptor.clone(); let held_processor_class = processor_class.clone().unbind(); // The closure captures the class's import path, never the class object: @@ -89,8 +94,8 @@ pub(crate) fn register_processor_class( let descriptor_for_constructor = declaration.descriptor.clone(); PROCESSOR_REGISTRY - .register_dynamic( - descriptor, + .install_constructor_for_registered_descriptor( + &identity, Box::new(move |node| { spawn_host_for_processor_node( &processor_class_import_path, @@ -104,12 +109,52 @@ pub(crate) fn register_processor_class( }) }), ) - .map_err(|registration_failure| PyValueError::new_err(registration_failure.to_string()))?; + .map_err(|install_failure| PyValueError::new_err(install_failure.to_string()))?; registered.insert(identity.clone(), held_processor_class); Ok(identity) } +/// Register the descriptor `@processor` has just stamped onto +/// `processor_class`, so the class is in the catalog before anything adds it. +/// +/// The decorator's one call into the native half. Registers the descriptor +/// alone: the constructor is the first add's to supply, through +/// [`register_processor_class`]. +/// +/// Two classes are passed over rather than registered. One decorated inside a +/// helper process registers nothing, because a helper hosts no graph. One no +/// interpreter could import — declared in the entry file or inside a function +/// — has no identity to be registered under, and `rt.add` is where that is +/// said, with the fix named. +#[pyfunction] +pub(crate) fn register_declared_processor_class(processor_class: &Bound<'_, PyAny>) -> PyResult<()> { + if std::env::var_os(HELPER_PROCESS_ENTRYPOINT_ENVIRONMENT_VARIABLE).is_some() { + return Ok(()); + } + if processor_class_import_path(processor_class).is_err() { + return Ok(()); + } + let declaration = PythonProcessorDeclaration::read_from_class(processor_class)?; + PROCESSOR_REGISTRY + .register_descriptor_only(declaration.descriptor) + .map_err(|registration_failure| PyValueError::new_err(registration_failure.to_string())) +} + +/// Every processor class import path the calling process has registered. +/// +/// The catalog `/api/registry` renders, reachable in a process that serves no +/// control plane — which a helper is, and is the only way to see from inside +/// one that decoration registered nothing there. +#[pyfunction] +pub(crate) fn processor_class_import_paths_registered_in_this_process() -> Vec { + PROCESSOR_REGISTRY + .list_registered() + .into_iter() + .map(|descriptor| descriptor.processor_class_import_path.as_str().to_string()) + .collect() +} + /// Register the class `processor_class_import_path` names by importing it into /// this interpreter — the registration import `rt.add` performs, done for a /// caller that holds only the path, such as an `add_processor` over the control diff --git a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py new file mode 100644 index 000000000..bbf1a7e10 --- /dev/null +++ b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py @@ -0,0 +1,169 @@ +# Copyright (c) 2025 Jonathan Fontanez +# SPDX-License-Identifier: BUSL-1.1 + +"""What `@processor` puts in the processor catalog the moment it runs. + +A class is discoverable before anything adds it, which is what lets an agent +read an app's effects off a node that has only imported them. No runtime boots +here: registration is a declaration-time fact, and the catalog is per process. +""" + +import sys +import types + +import pytest + +from streamlib import processor +from streamlib._engine import ( + processor_class_import_paths_registered_in_this_process, +) + + +@processor(execution="manual", description="Declared here and added nowhere") +class DeclaredAndNeverAdded: + """A processor this suite imports and never puts in a graph.""" + + +@processor(execution="manual") +class DescribedByItsDocstringAlone: + """What a processor with no description= keyword falls back to.""" + + +@processor(execution="manual", description="The keyword wins") +class DescribedByBothKeywordAndDocstring: + """The docstring the keyword outranks.""" + + +@processor(execution="manual") +class DescribedByNothingAtAll: + pass + + +def _declare_in_a_module_named(module_name: str, class_name: str) -> type: + """Run a decoration inside a module of the caller's naming. + + A test function's own classes carry `` in `__qualname__` and are + unimportable by design, so a module is the only place a decoration gets a + real import path — and a name per test is what keeps the process-wide + catalog from carrying one test's registration into another's. + """ + module = types.ModuleType(module_name) + sys.modules[module_name] = module + source = ( + "from streamlib import processor\n" + "@processor(execution='manual')\n" + f"class {class_name}:\n" + " pass\n" + ) + exec(compile(source, f"<{module_name}>", "exec"), module.__dict__) # noqa: S102 + return getattr(module, class_name) + + +def test_a_decorated_class_is_in_the_catalog_before_anything_adds_it(): + """The whole point: importing the module is the registration.""" + assert ( + "test_declaration_registers:DeclaredAndNeverAdded" + in processor_class_import_paths_registered_in_this_process() + ) + + +def test_the_catalog_names_a_class_by_its_import_path(): + """The same string a helper process imports the class back by.""" + declared = _declare_in_a_module_named( + "a_module_declaring_one_processor", "RegisteredAtDecoration" + ) + + assert declared.__module__ == "a_module_declaring_one_processor" + assert ( + "a_module_declaring_one_processor:RegisteredAtDecoration" + in processor_class_import_paths_registered_in_this_process() + ) + + +def test_a_class_declared_inside_a_function_registers_nothing(): + """It has no import path to be registered under. + + `rt.add` is where a class no interpreter can import is refused, with the + fix named — moving that refusal to decoration would refuse at import what + the plan refuses at add. + """ + catalog_before = set(processor_class_import_paths_registered_in_this_process()) + + @processor(execution="manual") + class DeclaredInsideThisTest: + pass + + assert "" in DeclaredInsideThisTest.__qualname__ + assert ( + set(processor_class_import_paths_registered_in_this_process()) + == catalog_before + ) + + +def test_one_import_path_decorated_twice_is_refused_naming_the_reload(): + """A module loaded twice rebuilds its classes, and both claim one path. + + The registry's duplicate refusal, now met at import where `importlib.reload` + is the cause a reader can act on. + """ + module = types.ModuleType("a_module_loaded_twice") + sys.modules["a_module_loaded_twice"] = module + source = compile( + "from streamlib import processor\n" + "@processor(execution='manual')\n" + "class DecoratedTwice:\n" + " pass\n", + "", + "exec", + ) + + exec(source, module.__dict__) # noqa: S102 + + with pytest.raises(ValueError) as refusal: + exec(source, module.__dict__) # noqa: S102 + + assert "a_module_loaded_twice:DecoratedTwice" in str(refusal.value) + assert "importlib.reload" in str(refusal.value) + + +def test_a_refused_second_decoration_leaves_the_first_registration_standing(): + """The registration that arrived first stays; nothing is overwritten.""" + _declare_in_a_module_named("a_module_reloaded_once", "SurvivesTheReload") + + module = sys.modules["a_module_reloaded_once"] + source = compile( + "from streamlib import processor\n" + "@processor(execution='manual')\n" + "class SurvivesTheReload:\n" + " pass\n", + "", + "exec", + ) + with pytest.raises(ValueError): + exec(source, module.__dict__) # noqa: S102 + + registered = processor_class_import_paths_registered_in_this_process() + assert ( + registered.count("a_module_reloaded_once:SurvivesTheReload") == 1 + ), "a refused duplicate must neither displace the first nor register beside it" + + +def test_a_processor_with_no_description_is_described_by_its_docstring(): + """The text an author already wrote, rather than a second place to write it.""" + assert ( + DescribedByItsDocstringAlone.__streamlib_processor_description__ + == "What a processor with no description= keyword falls back to." + ) + + +def test_an_explicit_description_outranks_the_docstring(): + """The keyword is the deliberate one; the docstring is the fallback.""" + assert ( + DescribedByBothKeywordAndDocstring.__streamlib_processor_description__ + == "The keyword wins" + ) + + +def test_a_processor_with_neither_is_described_by_the_empty_string(): + """Never `None`: the descriptor's description is a string.""" + assert DescribedByNothingAtAll.__streamlib_processor_description__ == "" From 82e3d8e21f5100b44ee710d33922122d1d547603 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:12:07 -0400 Subject: [PATCH 03/11] test(wheel): the catalog carries an imported class, and no helper's does Three proofs of declaration-time registration: a class the app imported and never added is served by `/api/registry` with its schema and its docstring description; a real helper hosting that class registers none of its module's declarations; and an interpreter carrying the helper's entrypoint variable registers nothing, which needs no device. Co-Authored-By: Claude Opus 5 --- .../tests/helper_placement_app.py | 23 ++++++++ .../tests/helper_placement_processors.py | 28 +++++++++ .../tests/processor_config_catalog_app.py | 8 ++- .../tests/processor_config_catalog_probes.py | 15 ++++- .../tests/test_declaration_registers.py | 58 +++++++++++++++++++ .../tests/test_helper_placement.py | 38 ++++++++++++ .../tests/test_processor_config_catalog.py | 38 +++++++++++- 7 files changed, 202 insertions(+), 6 deletions(-) diff --git a/sdk/streamlib-python-wheel/tests/helper_placement_app.py b/sdk/streamlib-python-wheel/tests/helper_placement_app.py index e85b1be44..c05b01b60 100644 --- a/sdk/streamlib-python-wheel/tests/helper_placement_app.py +++ b/sdk/streamlib-python-wheel/tests/helper_placement_app.py @@ -15,10 +15,14 @@ import streamlib from helper_placement_processors import ( DiesAbruptlyProbe, + ReportsItsOwnProcessesProcessorCatalog, ReportsItsOwnProcessSource, ReportsItsOwnProcessVideoSink, ReportsUpstreamProcessSink, ) +from streamlib._engine import ( + processor_class_import_paths_registered_in_this_process, +) MARKER_PREFIX = "MARKER:" @@ -152,5 +156,24 @@ def scenario_a_crashed_helper_leaves_the_pipeline_running() -> None: marker("CLEAN_EXIT") +def scenario_a_helper_registers_nothing_it_imports() -> None: + """The class is in the app's catalog from its import, and in no child's. + + The app side is the decorator's whole point — the class is discoverable + without ever being added. The child side is the other half: it imports the + same module to host the class and must register nothing, because a helper + hosts no graph. + """ + marker( + f"APP_CATALOG_HAS_THE_CLASS=" + f"{'helper_placement_processors:ReportsItsOwnProcessesProcessorCatalog' in processor_class_import_paths_registered_in_this_process()}" + ) + runtime = streamlib.Runtime() + runtime.add(ReportsItsOwnProcessesProcessorCatalog) + marker(f"APP_PID={os.getpid()}") + runtime.run() + marker("CLEAN_EXIT") + + if __name__ == "__main__": globals()[f"scenario_{sys.argv[1]}"]() diff --git a/sdk/streamlib-python-wheel/tests/helper_placement_processors.py b/sdk/streamlib-python-wheel/tests/helper_placement_processors.py index b22a444f9..3050bf48d 100644 --- a/sdk/streamlib-python-wheel/tests/helper_placement_processors.py +++ b/sdk/streamlib-python-wheel/tests/helper_placement_processors.py @@ -13,6 +13,9 @@ import os from streamlib import input, log, output, processor +from streamlib._engine import ( + processor_class_import_paths_registered_in_this_process, +) @dataclasses.dataclass @@ -102,3 +105,28 @@ def process(self, ctx) -> None: # Not an exception: the point is a process that stops existing # without unwinding, which is what a segfaulting native call does. os._exit(1) + + +@processor(execution="manual") +class ReportsItsOwnProcessesProcessorCatalog: + """Announces the processor catalog of the process it was constructed in. + + A helper hosts no graph, so importing this module inside one must leave its + registry empty of every class the module declares — the one thing the app + process cannot see for itself, because the registry is per process. The + count is of this module's classes rather than of the whole catalog: the + native built-ins register when the wheel's extension module initialises, + wherever that happens, and none of them came from a decorator. + """ + + def setup(self, ctx) -> None: + declared_by_this_module = [ + path + for path in processor_class_import_paths_registered_in_this_process() + if path.startswith("helper_placement_processors:") + ] + log.info( + f"MARKER:CHILD_CATALOG {os.getpid()} " + f"{'helper_placement_processors:ReportsItsOwnProcessesProcessorCatalog' in declared_by_this_module} " + f"{len(declared_by_this_module)}" + ) diff --git a/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py b/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py index 0af4ba084..7b379f209 100644 --- a/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py +++ b/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py @@ -5,7 +5,9 @@ Run as a real `python app.py`: four processors are added, each hosted in its own helper process, and this app then reads `GET /api/registry` off itself — the -exact payload an agent gets before deciding which keys a processor takes. +exact payload an agent gets before deciding which keys a processor takes. A +fifth class is imported and never added, which is what an agent discovering an +app's effects reads. """ import json @@ -34,6 +36,8 @@ def main() -> None: runtime.add(probes.DataclassConfiguredProbe, config={"width": 640, "label": "left"}) runtime.add(probes.ModelConfiguredProbe, config={"width": 1280}) runtime.add(probes.UnconfiguredProbe) + # `probes.ImportedButNeverAddedProbe` is deliberately not added: importing + # the module is what put it in the catalog. def read_the_catalog_this_node_serves() -> None: try: @@ -45,7 +49,7 @@ def read_the_catalog_this_node_serves() -> None: with urllib.request.urlopen(registry_url, timeout=READ_TIMEOUT_SECONDS) as response: served = json.load(response) catalog = { - entry["processor_class_import_path"]: entry.get("config_schema") + entry["processor_class_import_path"]: entry for entry in served["processors"] if entry["processor_class_import_path"].startswith( "processor_config_catalog_probes:" diff --git a/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py b/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py index 1177043ad..da700058c 100644 --- a/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py +++ b/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py @@ -1,13 +1,14 @@ # Copyright (c) 2025 Jonathan Fontanez # SPDX-License-Identifier: BUSL-1.1 -"""Four processors — three kinds of config class and one declaring none. +"""Five processors — three kinds of config class, one declaring none, one never added. Their own module, not the test module, because each runs in a helper process that reaches its class by importing the module it was declared in. -Each reports, from inside its helper, the object its `__init__` was handed — -which is the half of the contract a served schema cannot show. +Each added one reports, from inside its helper, the object its `__init__` was +handed — which is the half of the contract a served schema cannot show. The +fifth is added nowhere and reaches the catalog on its import alone. """ import dataclasses @@ -73,3 +74,11 @@ def setup(self, ctx: RuntimeContextFullAccess) -> None: class UnconfiguredProbe: def setup(self, ctx: RuntimeContextFullAccess) -> None: _report("UnconfiguredProbe", None) + + +@processor(execution="manual") +class ImportedButNeverAddedProbe: + """An effect the app knows how to run and has not been asked to.""" + + def __init__(self, config: DataclassProbeConfig) -> None: + self.config = config diff --git a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py index bbf1a7e10..ad21654f5 100644 --- a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py +++ b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py @@ -8,8 +8,12 @@ here: registration is a declaration-time fact, and the catalog is per process. """ +import json +import os +import subprocess import sys import types +from pathlib import Path import pytest @@ -18,6 +22,13 @@ processor_class_import_paths_registered_in_this_process, ) +HELPER_PROCESS_ENTRYPOINT_ENV = "STREAMLIB_ENTRYPOINT" + +# What a real helper imports and hosts — its own module, as every processor +# class must be. +PROCESSOR_MODULE_A_HELPER_HOSTS = "zero_argument_process_processor" +PROCESSOR_A_HELPER_HOSTS = f"{PROCESSOR_MODULE_A_HELPER_HOSTS}:ZeroArgumentProcess" + @processor(execution="manual", description="Declared here and added nowhere") class DeclaredAndNeverAdded: @@ -167,3 +178,50 @@ def test_an_explicit_description_outranks_the_docstring(): def test_a_processor_with_neither_is_described_by_the_empty_string(): """Never `None`: the descriptor's description is a string.""" assert DescribedByNothingAtAll.__streamlib_processor_description__ == "" + + +def _catalog_of_an_interpreter_carrying(environment: "dict[str, str]") -> "list[str]": + """The registry of a fresh interpreter that imported one processor module. + + Out of process because the variable under test is read once per import and + the catalog is per process: neither can be faked by patching inside this one. + """ + reporter = ( + f"import {PROCESSOR_MODULE_A_HELPER_HOSTS}\n" + "import json\n" + "from streamlib._engine import " + "processor_class_import_paths_registered_in_this_process as registered\n" + "print(json.dumps(registered()))\n" + ) + reported = subprocess.run( + [sys.executable, "-c", reporter], + check=True, + capture_output=True, + text=True, + env={ + **os.environ, + "PYTHONPATH": str(Path(__file__).parent), + **environment, + }, + ) + return json.loads(reported.stdout) + + +def test_an_interpreter_that_is_not_a_helper_registers_what_it_imports(): + """The control arm: the same import, without the helper's variable set.""" + assert PROCESSOR_A_HELPER_HOSTS in _catalog_of_an_interpreter_carrying({}) + + +def test_an_interpreter_carrying_the_helper_entrypoint_registers_nothing(): + """A helper hosts no graph, so it needs no catalog and builds none. + + The variable is the spawn host's, set on every child it starts and nowhere + else — which is what makes its presence a reliable "I am a helper". + """ + catalog = _catalog_of_an_interpreter_carrying( + {HELPER_PROCESS_ENTRYPOINT_ENV: PROCESSOR_A_HELPER_HOSTS} + ) + + assert PROCESSOR_A_HELPER_HOSTS not in catalog, ( + f"a helper registered the class it hosts: {catalog}" + ) diff --git a/sdk/streamlib-python-wheel/tests/test_helper_placement.py b/sdk/streamlib-python-wheel/tests/test_helper_placement.py index ff2b2d950..0c3645ca1 100644 --- a/sdk/streamlib-python-wheel/tests/test_helper_placement.py +++ b/sdk/streamlib-python-wheel/tests/test_helper_placement.py @@ -35,6 +35,7 @@ VIDEO_SINK_PID_MARKER = re.compile(r"MARKER:VIDEO_SINK_PID (\d+)") APP_PID_MARKER = re.compile(r"MARKER:APP_PID=(\d+)") HELPER_STARTED_MARKER = re.compile(r"helper process started: pid=(\d+)") +CHILD_CATALOG_MARKER = re.compile(r"MARKER:CHILD_CATALOG (\d+) (True|False) (\d+)") def run_scenario(start_app_under_test, scenario: str): @@ -180,6 +181,43 @@ def test_a_native_builtin_stays_in_the_app_process(start_app_under_test): ) +def test_a_helper_registers_nothing_the_module_it_imports_declares(start_app_under_test): + """The app's catalog carries the class; the child that hosts it carries none. + + A helper hosts no graph, so the registry a decoration would fill has no + reader there. The child imports the same module the app did — that import + is how it reaches the class at all — so the two catalogs are the whole + difference, and the child reports its own. + """ + app = run_scenario_until( + start_app_under_test, + "a_helper_registers_nothing_it_imports", + "MARKER:CHILD_CATALOG", + "the helper to report the catalog of the process it was constructed in", + ) + + assert "APP_CATALOG_HAS_THE_CLASS=True" in app.output, ( + f"the app never registered a class it imported:\n{app.output}" + ) + + app_pid = int(matched_marker(APP_PID_MARKER, app.output).group(1)) + child_pid, child_registered_the_class, declared_classes_in_the_child = ( + matched_marker(CHILD_CATALOG_MARKER, app.output).groups() + ) + + assert int(child_pid) != app_pid, ( + f"the processor reported the app's own process ({app_pid}), so this " + f"says nothing about a helper:\n{app.output}" + ) + assert child_registered_the_class == "False", ( + f"the helper registered the class it hosts:\n{app.output}" + ) + assert declared_classes_in_the_child == "0", ( + f"the helper registered {declared_classes_in_the_child} of the module's " + f"declared classes — a helper hosts no graph:\n{app.output}" + ) + + def helper_process_is_still_alive(pid: int) -> bool: """Whether `pid` is still a live streamlib helper. diff --git a/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py b/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py index f2eda8b05..f7470128c 100644 --- a/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py +++ b/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py @@ -53,12 +53,48 @@ def served_catalog(catalog_app_output: str) -> "dict[str, Any]": return json.loads(match.group(1)) +def entry_for(served_catalog: "dict[str, Any]", probe: str) -> "dict[str, Any]": + return served_catalog[f"processor_config_catalog_probes:{probe}"] + + def schema_for(served_catalog: "dict[str, Any]", probe: str) -> "dict[str, Any]": - document = served_catalog[f"processor_config_catalog_probes:{probe}"] + document = entry_for(served_catalog, probe).get("config_schema") assert document is not None, f"{probe} served a null config schema" return document +def test_a_class_the_app_imported_and_never_added_is_in_the_catalog(served_catalog): + """What an agent reads to learn what a node could run, not what it is running. + + Its decorator registered it at import; nothing put it in the graph. Restore + registration to the first add and it is invisible here, which is the gap + that made an app's unused effects undiscoverable. + """ + entry = entry_for(served_catalog, "ImportedButNeverAddedProbe") + + assert entry["config_schema"]["properties"]["width"]["type"] == "integer", ( + f"an unadded class must carry the same config schema an added one does: {entry}" + ) + assert entry["runtime"] == "python" + assert entry["entrypoint"] == ( + "processor_config_catalog_probes:ImportedButNeverAddedProbe" + ) + + +def test_a_processor_with_no_description_is_served_its_docstring(served_catalog): + """The text the author already wrote reaches the agent reading the catalog.""" + assert entry_for(served_catalog, "ImportedButNeverAddedProbe")["description"] == ( + "An effect the app knows how to run and has not been asked to." + ) + + +def test_an_explicit_description_is_served_over_the_docstring(served_catalog): + assert ( + entry_for(served_catalog, "UnconfiguredProbe")["description"] + == "Takes no configuration at all" + ) + + def test_a_dataclass_config_reaches_the_registry_with_types_defaults_and_descriptions( served_catalog, ): From e0961df9155ff47b5f6f7db7f361dd98fae71b98 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:18:35 -0400 Subject: [PATCH 04/11] test(wheel): the embedded decorator harness stands in for _engine too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The decorator now imports the native half, and the harness's search path points at the source directory — where a `maturin develop` leaves a compiled engine with its own process-global registry. A stand-in module keeps these tests reading the grammar and registering nothing, and keeps them running where no artifact was built. Co-Authored-By: Claude Opus 5 --- .../src/python_processor_declaration.rs | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs index cf2bfade3..db5e7b49f 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs @@ -517,6 +517,13 @@ class BlurProcessor: /// `__init__.py` — which imports the compiled `_engine` a `cargo test` run /// does not have — is not executed. Only the siblings actually imported are /// loaded, and each is the same file `include_str!` above reads. + /// + /// `_engine` is the one sibling that cannot be: it is the compiled artifact + /// this binary *is* a copy of, and a `maturin develop` leaves one in the + /// source directory that the search path would otherwise load — a second + /// engine, with its own process-global registry, deciding whether these + /// tests pass. A stand-in stands in for it, so a decoration here reads the + /// grammar and registers nothing. fn install_stand_in_streamlib_package(python: Python<'_>) { let sys_modules = python .import("sys") @@ -541,6 +548,32 @@ class BlurProcessor: ) .unwrap(); sys_modules.set_item("streamlib", package).unwrap(); + + let stand_in_engine = python + .import("types") + .unwrap() + .call_method1("ModuleType", ("streamlib._engine",)) + .unwrap(); + let namespace = PyDict::new(python); + python + .run( + c"def register_declared_processor_class(processor_class): pass", + Some(&namespace), + None, + ) + .unwrap(); + stand_in_engine + .setattr( + "register_declared_processor_class", + namespace + .get_item("register_declared_processor_class") + .unwrap() + .unwrap(), + ) + .unwrap(); + sys_modules + .set_item("streamlib._engine", stand_in_engine) + .unwrap(); } /// Run the real decorator module, run `class_body_source` against it, and From 46c3393b13b18004326d7b198161a0058a2c0886 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:19:21 -0400 Subject: [PATCH 05/11] docs(wheel): register_processor_class installs, and its doc says so Co-Authored-By: Claude Opus 5 --- .../src/python_processor_registration.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdk/streamlib-python-wheel/src/python_processor_registration.rs b/sdk/streamlib-python-wheel/src/python_processor_registration.rs index 43462874f..c011440df 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_registration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_registration.rs @@ -36,7 +36,8 @@ fn registered_processor_classes() -> &'static Mutex Date: Fri, 11 Sep 2026 17:20:15 -0400 Subject: [PATCH 06/11] fix(wheel): the stub's __all__ names the two new engine functions stubtest compares the stub's exports to the binary's; a new pyfunction is not done until both the entry and the export list carry it. Co-Authored-By: Claude Opus 5 --- sdk/streamlib-python-wheel/python/streamlib/_engine.pyi | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index e1aa228ac..e88f0f4f4 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi +++ b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi @@ -77,6 +77,8 @@ __all__ = [ "log_event", "monotonic_now_ns", "open_test_harness_channel", + "processor_class_import_paths_registered_in_this_process", + "register_declared_processor_class", "runtime_log_directory", ] From 0a77a24ef9d0c9dd5c6c3159dc7fd298edacee9e Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:22:02 -0400 Subject: [PATCH 07/11] style(wheel): rustfmt the new registration entry Co-Authored-By: Claude Opus 5 --- .../src/python_processor_registration.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sdk/streamlib-python-wheel/src/python_processor_registration.rs b/sdk/streamlib-python-wheel/src/python_processor_registration.rs index c011440df..65d0d6576 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_registration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_registration.rs @@ -129,7 +129,9 @@ pub(crate) fn register_processor_class( /// — has no identity to be registered under, and `rt.add` is where that is /// said, with the fix named. #[pyfunction] -pub(crate) fn register_declared_processor_class(processor_class: &Bound<'_, PyAny>) -> PyResult<()> { +pub(crate) fn register_declared_processor_class( + processor_class: &Bound<'_, PyAny>, +) -> PyResult<()> { if std::env::var_os(HELPER_PROCESS_ENTRYPOINT_ENVIRONMENT_VARIABLE).is_some() { return Ok(()); } From 8d3797288bb38aad9e28f926b9ae79edddbc27ed Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:32:42 -0400 Subject: [PATCH 08/11] fix(engine): the unknown-path refusal reads as a sentence, not a gutter A source-wrapped literal written without its line continuations carried eighteen spaces of indentation into the message a Python author sees. The test that names the path now also refuses a gutter in the prose. Beside it, review follow-ups: the catalog projection reads the registry's keys rather than cloning every descriptor to discard it; the wheel's listing is named for the catalog it reads, since "registered" already means "has a constructor" on the engine's side; both decoration-time skips say so at debug; and the embedded harness builds its stand-in module in one call. Co-Authored-By: Claude Opus 5 --- .../processors/processor_instance_factory.rs | 18 ++++++++++- .../python/streamlib/_engine.pyi | 11 +++---- sdk/streamlib-python-wheel/src/lib.rs | 2 +- .../src/python_processor_declaration.rs | 29 +++++------------- .../src/python_processor_registration.rs | 30 ++++++++++++++----- .../tests/helper_placement_app.py | 4 +-- .../tests/helper_placement_processors.py | 4 +-- .../tests/test_declaration_registers.py | 14 ++++----- 8 files changed, 64 insertions(+), 48 deletions(-) diff --git a/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs b/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs index cdaeac58c..41989a656 100644 --- a/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs +++ b/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs @@ -455,7 +455,10 @@ impl ProcessorInstanceFactory { let descriptors = self.descriptors.read(); if !descriptors.contains_key(processor_class_import_path) { return Err(Error::ProcessorNotFound(format!( - "no descriptor is registered for processor type '{processor_class_import_path}', so there is nothing to install a constructor onto. A Python class registers its descriptor when its `@processor` decorator runs, so a path missing here names a class this process never imported." + "no descriptor is registered for processor type \ + '{processor_class_import_path}', so there is nothing to install a constructor \ + onto. A Python class registers its descriptor when its `@processor` decorator \ + runs, so a path missing here names a class this process never imported." ))); } @@ -614,6 +617,15 @@ impl ProcessorInstanceFactory { .map(|descriptor| descriptor.processor_class_short_name.as_str().to_string()) } + /// Every processor class import path the registry holds a descriptor for. + /// + /// Projects the keys out under the read lock rather than going through + /// [`Self::list_registered`], which clones every descriptor whole — port + /// vectors and config schema included — for callers that want the names. + pub fn registered_processor_class_import_paths(&self) -> Vec { + self.descriptors.read().keys().cloned().collect() + } + /// List all registered processor types with their full descriptors. pub fn list_registered(&self) -> Vec { self.descriptors.read().values().cloned().collect() @@ -929,6 +941,10 @@ mod tests { message.contains(path), "the refusal must name the path nobody registered; got: {message}" ); + assert!( + !message.contains(" "), + "a source-wrapped message must carry no gutter into its text; got: {message}" + ); assert!( factory.descriptor(&class_import_path(path)).is_none(), "a refused install must leave the registry untouched" diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index e88f0f4f4..efb7c7f7a 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi +++ b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi @@ -77,7 +77,7 @@ __all__ = [ "log_event", "monotonic_now_ns", "open_test_harness_channel", - "processor_class_import_paths_registered_in_this_process", + "processor_class_import_paths_in_this_processes_catalog", "register_declared_processor_class", "runtime_log_directory", ] @@ -1652,11 +1652,12 @@ def register_declared_processor_class(processor_class: type) -> None: interpreter could import, which `Runtime.add` refuses by name. """ -def processor_class_import_paths_registered_in_this_process() -> list[str]: - """Every processor class import path the calling process has registered. +def processor_class_import_paths_in_this_processes_catalog() -> list[str]: + """Every processor class import path in the calling process's catalog. - The catalog `GET /api/registry` renders, readable in a process that serves - no control plane. + What `GET /api/registry` renders, readable in a process that serves no + control plane. A path appears here from the moment its `@processor` + decorator runs, whether or not anything has added it. """ def monotonic_now_ns() -> int: diff --git a/sdk/streamlib-python-wheel/src/lib.rs b/sdk/streamlib-python-wheel/src/lib.rs index c3e66bb89..b83ae1793 100644 --- a/sdk/streamlib-python-wheel/src/lib.rs +++ b/sdk/streamlib-python-wheel/src/lib.rs @@ -105,7 +105,7 @@ fn _engine(module: &Bound<'_, PyModule>) -> PyResult<()> { module )?)?; module.add_function(wrap_pyfunction!( - python_processor_registration::processor_class_import_paths_registered_in_this_process, + python_processor_registration::processor_class_import_paths_in_this_processes_catalog, module )?)?; module.add_function(wrap_pyfunction!(python_logging::monotonic_now_ns, module)?)?; diff --git a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs index db5e7b49f..8e090cad0 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs @@ -549,28 +549,13 @@ class BlurProcessor: .unwrap(); sys_modules.set_item("streamlib", package).unwrap(); - let stand_in_engine = python - .import("types") - .unwrap() - .call_method1("ModuleType", ("streamlib._engine",)) - .unwrap(); - let namespace = PyDict::new(python); - python - .run( - c"def register_declared_processor_class(processor_class): pass", - Some(&namespace), - None, - ) - .unwrap(); - stand_in_engine - .setattr( - "register_declared_processor_class", - namespace - .get_item("register_declared_processor_class") - .unwrap() - .unwrap(), - ) - .unwrap(); + let stand_in_engine = PyModule::from_code( + python, + c"def register_declared_processor_class(processor_class): pass", + c"streamlib/_engine.py", + c"streamlib._engine", + ) + .unwrap(); sys_modules .set_item("streamlib._engine", stand_in_engine) .unwrap(); diff --git a/sdk/streamlib-python-wheel/src/python_processor_registration.rs b/sdk/streamlib-python-wheel/src/python_processor_registration.rs index 65d0d6576..7bac3af8f 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_registration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_registration.rs @@ -133,9 +133,21 @@ pub(crate) fn register_declared_processor_class( processor_class: &Bound<'_, PyAny>, ) -> PyResult<()> { if std::env::var_os(HELPER_PROCESS_ENTRYPOINT_ENVIRONMENT_VARIABLE).is_some() { + tracing::debug!( + "[register_declared_processor_class] a decoration inside a helper process registers \ + nothing" + ); return Ok(()); } - if processor_class_import_path(processor_class).is_err() { + // Only an unresolvable identity is passed over, and deliberately narrowly: + // every other refusal `read_from_class` raises — a malformed port, an + // unreadable config schema — is the author's to see at decoration, so this + // guard asks the one question rather than swallowing the whole read. + if let Err(no_import_path) = processor_class_import_path(processor_class) { + tracing::debug!( + %no_import_path, + "[register_declared_processor_class] a class with no import path registers nothing" + ); return Ok(()); } let declaration = PythonProcessorDeclaration::read_from_class(processor_class)?; @@ -144,17 +156,19 @@ pub(crate) fn register_declared_processor_class( .map_err(|registration_failure| PyValueError::new_err(registration_failure.to_string())) } -/// Every processor class import path the calling process has registered. +/// Every processor class import path in the calling process's catalog. /// -/// The catalog `/api/registry` renders, reachable in a process that serves no -/// control plane — which a helper is, and is the only way to see from inside -/// one that decoration registered nothing there. +/// What `/api/registry` renders, reachable in a process that serves no control +/// plane — which a helper is, and is the only way to see from inside one that +/// decoration registered nothing there. Named for the catalog rather than for +/// registration: a path listed here may be one the engine's `is_registered` +/// calls false, because that asks whether a constructor has arrived. #[pyfunction] -pub(crate) fn processor_class_import_paths_registered_in_this_process() -> Vec { +pub(crate) fn processor_class_import_paths_in_this_processes_catalog() -> Vec { PROCESSOR_REGISTRY - .list_registered() + .registered_processor_class_import_paths() .into_iter() - .map(|descriptor| descriptor.processor_class_import_path.as_str().to_string()) + .map(|import_path| import_path.as_str().to_string()) .collect() } diff --git a/sdk/streamlib-python-wheel/tests/helper_placement_app.py b/sdk/streamlib-python-wheel/tests/helper_placement_app.py index c05b01b60..8925250af 100644 --- a/sdk/streamlib-python-wheel/tests/helper_placement_app.py +++ b/sdk/streamlib-python-wheel/tests/helper_placement_app.py @@ -21,7 +21,7 @@ ReportsUpstreamProcessSink, ) from streamlib._engine import ( - processor_class_import_paths_registered_in_this_process, + processor_class_import_paths_in_this_processes_catalog, ) MARKER_PREFIX = "MARKER:" @@ -166,7 +166,7 @@ def scenario_a_helper_registers_nothing_it_imports() -> None: """ marker( f"APP_CATALOG_HAS_THE_CLASS=" - f"{'helper_placement_processors:ReportsItsOwnProcessesProcessorCatalog' in processor_class_import_paths_registered_in_this_process()}" + f"{'helper_placement_processors:ReportsItsOwnProcessesProcessorCatalog' in processor_class_import_paths_in_this_processes_catalog()}" ) runtime = streamlib.Runtime() runtime.add(ReportsItsOwnProcessesProcessorCatalog) diff --git a/sdk/streamlib-python-wheel/tests/helper_placement_processors.py b/sdk/streamlib-python-wheel/tests/helper_placement_processors.py index 3050bf48d..a86770d3b 100644 --- a/sdk/streamlib-python-wheel/tests/helper_placement_processors.py +++ b/sdk/streamlib-python-wheel/tests/helper_placement_processors.py @@ -14,7 +14,7 @@ from streamlib import input, log, output, processor from streamlib._engine import ( - processor_class_import_paths_registered_in_this_process, + processor_class_import_paths_in_this_processes_catalog, ) @@ -122,7 +122,7 @@ class ReportsItsOwnProcessesProcessorCatalog: def setup(self, ctx) -> None: declared_by_this_module = [ path - for path in processor_class_import_paths_registered_in_this_process() + for path in processor_class_import_paths_in_this_processes_catalog() if path.startswith("helper_placement_processors:") ] log.info( diff --git a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py index ad21654f5..03cd8e1f9 100644 --- a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py +++ b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py @@ -19,7 +19,7 @@ from streamlib import processor from streamlib._engine import ( - processor_class_import_paths_registered_in_this_process, + processor_class_import_paths_in_this_processes_catalog, ) HELPER_PROCESS_ENTRYPOINT_ENV = "STREAMLIB_ENTRYPOINT" @@ -74,7 +74,7 @@ def test_a_decorated_class_is_in_the_catalog_before_anything_adds_it(): """The whole point: importing the module is the registration.""" assert ( "test_declaration_registers:DeclaredAndNeverAdded" - in processor_class_import_paths_registered_in_this_process() + in processor_class_import_paths_in_this_processes_catalog() ) @@ -87,7 +87,7 @@ def test_the_catalog_names_a_class_by_its_import_path(): assert declared.__module__ == "a_module_declaring_one_processor" assert ( "a_module_declaring_one_processor:RegisteredAtDecoration" - in processor_class_import_paths_registered_in_this_process() + in processor_class_import_paths_in_this_processes_catalog() ) @@ -98,7 +98,7 @@ def test_a_class_declared_inside_a_function_registers_nothing(): fix named — moving that refusal to decoration would refuse at import what the plan refuses at add. """ - catalog_before = set(processor_class_import_paths_registered_in_this_process()) + catalog_before = set(processor_class_import_paths_in_this_processes_catalog()) @processor(execution="manual") class DeclaredInsideThisTest: @@ -106,7 +106,7 @@ class DeclaredInsideThisTest: assert "" in DeclaredInsideThisTest.__qualname__ assert ( - set(processor_class_import_paths_registered_in_this_process()) + set(processor_class_import_paths_in_this_processes_catalog()) == catalog_before ) @@ -153,7 +153,7 @@ def test_a_refused_second_decoration_leaves_the_first_registration_standing(): with pytest.raises(ValueError): exec(source, module.__dict__) # noqa: S102 - registered = processor_class_import_paths_registered_in_this_process() + registered = processor_class_import_paths_in_this_processes_catalog() assert ( registered.count("a_module_reloaded_once:SurvivesTheReload") == 1 ), "a refused duplicate must neither displace the first nor register beside it" @@ -190,7 +190,7 @@ def _catalog_of_an_interpreter_carrying(environment: "dict[str, str]") -> "list[ f"import {PROCESSOR_MODULE_A_HELPER_HOSTS}\n" "import json\n" "from streamlib._engine import " - "processor_class_import_paths_registered_in_this_process as registered\n" + "processor_class_import_paths_in_this_processes_catalog as registered\n" "print(json.dumps(registered()))\n" ) reported = subprocess.run( From cdb7477176dbad0183fc93a2bd5c5740d46a9184 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:37:08 -0400 Subject: [PATCH 09/11] docs: the claims this change falsified now say what is true MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `register_descriptor_only` is the decorator's door, and `create()` on one of its paths succeeds as soon as the first add installs the constructor — its doc said neither. The decorator module said the native half reads its attributes at add time; it reads them at decoration. The wheel's class cache guards installs, not registrations. The helper-entrypoint test reads the variable's name from `_helper` rather than spelling it a third time. Co-Authored-By: Claude Opus 5 --- .../core/processors/processor_instance_factory.rs | 13 ++++++------- .../python/streamlib/_engine.pyi | 8 ++++++-- .../python/streamlib/_processor_declaration.py | 8 +++++--- .../src/python_processor_registration.rs | 8 ++++---- .../tests/test_declaration_registers.py | 5 ++--- 5 files changed, 23 insertions(+), 19 deletions(-) diff --git a/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs b/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs index 41989a656..d95b8241c 100644 --- a/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs +++ b/runtime/streamlib-engine/src/core/processors/processor_instance_factory.rs @@ -396,9 +396,11 @@ impl ProcessorInstanceFactory { /// Register a processor descriptor without a constructor. /// - /// Used for subprocess processors (Python, TypeScript) where no Rust-side - /// `ProcessorInstance` is created. The graph needs the descriptor and port info - /// for validation and wiring, but `create()` will return an error if called. + /// What a Python class's `@processor` decorator calls, so the class is in + /// the catalog before anything adds it. The graph has the descriptor and + /// port info it needs to validate and wire, and `create()` refuses until + /// [`Self::install_constructor_for_registered_descriptor`] supplies the + /// constructor — which a first add does. pub fn register_descriptor_only(&self, descriptor: ProcessorDescriptor) -> Result<()> { let processor_class_import_path = descriptor.processor_class_import_path.clone(); @@ -418,11 +420,8 @@ impl ProcessorInstanceFactory { descriptors.insert(processor_class_import_path.clone(), descriptor); drop(descriptors); - // No constructor registered - create() will fail with ProcessorNotFound, - // which is correct since subprocess processors are never instantiated in Rust. - tracing::info!( - "[register_descriptor_only] subprocess processor type registered '{}'", + "[register_descriptor_only] processor type registered without a constructor '{}'", processor_class_import_path ); diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index efb7c7f7a..7404b02b5 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi +++ b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi @@ -1656,8 +1656,12 @@ def processor_class_import_paths_in_this_processes_catalog() -> list[str]: """Every processor class import path in the calling process's catalog. What `GET /api/registry` renders, readable in a process that serves no - control plane. A path appears here from the moment its `@processor` - decorator runs, whether or not anything has added it. + control plane — which a helper process is. A path appears here from the + moment its `@processor` decorator runs, whether or not anything has added + it, so a path listed here may be one the engine cannot yet construct. + + The wheel's own suites read it to see a helper's catalog from inside one; + an app reads the control plane instead. """ def monotonic_now_ns() -> int: diff --git a/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py b/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py index 9d4ae62c4..ead8aa233 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py +++ b/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py @@ -4,9 +4,11 @@ """The `@processor` grammar — execution mode and ports, declared in code. Nothing is read from disk: there is no manifest, and a bare `.py` module defines -a working processor. `@processor` attaches the metadata the engine reads at -`Runtime.add` time as `__streamlib_processor_*__` class attributes; that set is -the contract between this module and the native half, and the two move together. +a working processor. `@processor` attaches the metadata as +`__streamlib_processor_*__` class attributes and hands the class to the native +half, which reads exactly that set and registers the descriptor there and then; +the set is the contract between this module and the native half, and the two +move together. Ports are declared with the `@input` / `@output` method decorators and accessed at run time through `ctx.inputs` / `ctx.outputs` — the marker methods themselves are never called. diff --git a/sdk/streamlib-python-wheel/src/python_processor_registration.rs b/sdk/streamlib-python-wheel/src/python_processor_registration.rs index 7bac3af8f..ae8e97534 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_registration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_registration.rs @@ -24,11 +24,11 @@ use crate::python_helper_process_spawn_host::{ use crate::python_processor_declaration::PythonProcessorDeclaration; use crate::python_processor_import_path::processor_class_import_path; -/// Which Python class each registered import path was registered from. +/// Which Python class each import path had its constructor installed from. /// -/// A cache of *which class*, never the authority on *whether* a type is -/// registered — that stays the engine's registry, consulted below, so this can -/// never suppress a re-registration the engine actually needs. +/// A cache of *which class*, never the authority on *whether* a type can be +/// constructed — that stays the engine's registry, consulted below, so this can +/// never report an install the engine does not actually hold. fn registered_processor_classes() -> &'static Mutex>> { static REGISTERED_PROCESSOR_CLASSES: OnceLock< Mutex>>, diff --git a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py index 03cd8e1f9..49eb1c392 100644 --- a/sdk/streamlib-python-wheel/tests/test_declaration_registers.py +++ b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py @@ -21,8 +21,7 @@ from streamlib._engine import ( processor_class_import_paths_in_this_processes_catalog, ) - -HELPER_PROCESS_ENTRYPOINT_ENV = "STREAMLIB_ENTRYPOINT" +from streamlib._helper import ENTRYPOINT_ENV # What a real helper imports and hosts — its own module, as every processor # class must be. @@ -219,7 +218,7 @@ def test_an_interpreter_carrying_the_helper_entrypoint_registers_nothing(): else — which is what makes its presence a reliable "I am a helper". """ catalog = _catalog_of_an_interpreter_carrying( - {HELPER_PROCESS_ENTRYPOINT_ENV: PROCESSOR_A_HELPER_HOSTS} + {ENTRYPOINT_ENV: PROCESSOR_A_HELPER_HOSTS} ) assert PROCESSOR_A_HELPER_HOSTS not in catalog, ( From c3665e107a5e5e13112ce8c63a8fc31fb63d476c Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Fri, 11 Sep 2026 17:46:50 -0400 Subject: [PATCH 10/11] docs(wheel): the catalog listing's stub says which process it describes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The paragraph named a helper as the reason to call it and then stated the app-process rule as general, where the helper case is the opposite — and the suites it points at assert exactly that opposite. Co-Authored-By: Claude Opus 5 --- .../python/streamlib/_engine.pyi | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index 7404b02b5..5305af50e 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi +++ b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi @@ -1656,12 +1656,12 @@ def processor_class_import_paths_in_this_processes_catalog() -> list[str]: """Every processor class import path in the calling process's catalog. What `GET /api/registry` renders, readable in a process that serves no - control plane — which a helper process is. A path appears here from the - moment its `@processor` decorator runs, whether or not anything has added - it, so a path listed here may be one the engine cannot yet construct. - - The wheel's own suites read it to see a helper's catalog from inside one; - an app reads the control plane instead. + control plane — which a helper process is. In the app process a path + appears here the moment its `@processor` decorator runs, whether or not + anything has added it, so a path listed here may be one the engine cannot + yet construct. In a helper nothing appears, because decoration registers + nothing there — and seeing that from inside one is what the wheel's own + suites read this for. """ def monotonic_now_ns() -> int: From 44958c77146cb8eba3253227ec64aa0a400da137 Mon Sep 17 00:00:00 2001 From: Jonathan Fontanez Date: Sat, 12 Sep 2026 20:25:37 -0400 Subject: [PATCH 11/11] test(wheel): the stand-in claims each half of the package on its own The `_engine` stand-in this change added sits behind a guard that returns as soon as `streamlib` is on `sys.modules`, so a `streamlib` present without `streamlib._engine` skipped it and left the decorator module's relative import to find the compiled artifact this binary is a copy of. Each half is claimed separately now. Co-Authored-By: Claude Opus 5 --- .../src/python_processor_declaration.rs | 49 ++++++++++--------- 1 file changed, 26 insertions(+), 23 deletions(-) diff --git a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs index 8e090cad0..fc1d1d4a3 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs @@ -532,33 +532,36 @@ class BlurProcessor: .unwrap() .cast_into::() .unwrap(); - if sys_modules.contains("streamlib").unwrap() { - return; + // Each half is claimed on its own: a `streamlib` already on `sys.modules` without + // `streamlib._engine` would otherwise skip the stand-in and leave the relative import + // to find the compiled artifact this binary is a copy of. + if !sys_modules.contains("streamlib").unwrap() { + let package = python + .import("types") + .unwrap() + .call_method1("ModuleType", ("streamlib",)) + .unwrap(); + package + .setattr( + "__path__", + PyList::new(python, [WHEEL_PYTHON_PACKAGE_DIRECTORY]).unwrap(), + ) + .unwrap(); + sys_modules.set_item("streamlib", package).unwrap(); } - let package = python - .import("types") - .unwrap() - .call_method1("ModuleType", ("streamlib",)) - .unwrap(); - package - .setattr( - "__path__", - PyList::new(python, [WHEEL_PYTHON_PACKAGE_DIRECTORY]).unwrap(), + if !sys_modules.contains("streamlib._engine").unwrap() { + let stand_in_engine = PyModule::from_code( + python, + c"def register_declared_processor_class(processor_class): pass", + c"streamlib/_engine.py", + c"streamlib._engine", ) .unwrap(); - sys_modules.set_item("streamlib", package).unwrap(); - - let stand_in_engine = PyModule::from_code( - python, - c"def register_declared_processor_class(processor_class): pass", - c"streamlib/_engine.py", - c"streamlib._engine", - ) - .unwrap(); - sys_modules - .set_item("streamlib._engine", stand_in_engine) - .unwrap(); + sys_modules + .set_item("streamlib._engine", stand_in_engine) + .unwrap(); + } } /// Run the real decorator module, run `class_body_source` against it, and