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..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 ); @@ -436,6 +435,51 @@ 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, @@ -572,6 +616,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() @@ -795,6 +848,108 @@ 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!( + !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" + ); + } + #[test] fn an_unregistered_path_is_absent_from_every_map() { let factory = ProcessorInstanceFactory::new(); diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index 8745b3abc..5305af50e 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_in_this_processes_catalog", + "register_declared_processor_class", "runtime_log_directory", ] @@ -1640,6 +1642,28 @@ 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_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. 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: """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..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. @@ -22,6 +24,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 +358,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 +414,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 +425,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..b83ae1793 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_in_this_processes_catalog, + 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_declaration.rs b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs index cf2bfade3..fc1d1d4a3 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") @@ -525,22 +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(); + sys_modules + .set_item("streamlib._engine", stand_in_engine) + .unwrap(); + } } /// Run the real decorator module, run `class_body_source` against it, and diff --git a/sdk/streamlib-python-wheel/src/python_processor_registration.rs b/sdk/streamlib-python-wheel/src/python_processor_registration.rs index d8f62a919..ae8e97534 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,14 +18,17 @@ 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. +/// 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>>, @@ -30,7 +36,8 @@ fn registered_processor_classes() -> &'static Mutex, +) -> 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(()); + } + // 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)?; + PROCESSOR_REGISTRY + .register_descriptor_only(declaration.descriptor) + .map_err(|registration_failure| PyValueError::new_err(registration_failure.to_string())) +} + +/// Every processor class import path in the calling process's catalog. +/// +/// 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_in_this_processes_catalog() -> Vec { + PROCESSOR_REGISTRY + .registered_processor_class_import_paths() + .into_iter() + .map(|import_path| 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/helper_placement_app.py b/sdk/streamlib-python-wheel/tests/helper_placement_app.py index e85b1be44..8925250af 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_in_this_processes_catalog, +) 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_in_this_processes_catalog()}" + ) + 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..a86770d3b 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_in_this_processes_catalog, +) @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_in_this_processes_catalog() + 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 new file mode 100644 index 000000000..49eb1c392 --- /dev/null +++ b/sdk/streamlib-python-wheel/tests/test_declaration_registers.py @@ -0,0 +1,226 @@ +# 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 json +import os +import subprocess +import sys +import types +from pathlib import Path + +import pytest + +from streamlib import processor +from streamlib._engine import ( + processor_class_import_paths_in_this_processes_catalog, +) +from streamlib._helper import ENTRYPOINT_ENV + +# 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: + """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_in_this_processes_catalog() + ) + + +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_in_this_processes_catalog() + ) + + +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_in_this_processes_catalog()) + + @processor(execution="manual") + class DeclaredInsideThisTest: + pass + + assert "" in DeclaredInsideThisTest.__qualname__ + assert ( + set(processor_class_import_paths_in_this_processes_catalog()) + == 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_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" + + +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__ == "" + + +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_in_this_processes_catalog 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( + {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, ): 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",