feat(wheel)!: @processor registers the descriptor when it runs; the constructor arrives at first add

.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 \

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<ProcessorClassImportPath> {
+        self.descriptors.read().keys().cloned().collect()
+    }
+
     /// List all registered processor types with their full descriptors.
     pub fn list_registered(&self) -> Vec<ProcessorDescriptor> {
         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();

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)`."""
 

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
 
 

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!(

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(