feat(engine)!: the descriptor carries the config type's derived JSON Schema

.github/workflows/test.yml

@@ -121,6 +121,14 @@ jobs:
           cargo test --locked -p streamlib -p streamlib-macros \
             -p streamlib-processor-schema --lib
           cargo test --locked -p streamlib-engine --test attribute_macro_test
+          # The compile-fail half of the same locks. The descriptor carries the
+          # config type's schema, so the derive is a requirement, and the note
+          # naming the fix is the only thing standing between an author and a
+          # bare trait-bound error. Nothing else in CI hands the compiler a
+          # config type without the derive. A compiler upgrade that reflows a
+          # diagnostic reddens here; `TRYBUILD=overwrite` refreshes it.
+          cargo test --locked -p streamlib-engine \
+            --test compile_fail_config_without_json_schema
           # The built-ins were reached by no cargo invocation here: the clippy
           # step above compiles default targets only, so it builds their lib and
           # never their test target. Their unit tests are the headless arm of
@@ -195,6 +203,11 @@ jobs:
           # builder writes, so moving one without the other reddens here rather
           # than on the rig. It gates the two values agreeing, not the emitted
           # VUI — the dB that a disagreement costs is rig-measured.
+          # The config-schema entries are the only gate the descriptor's
+          # config slot has on the way to the control plane: that the
+          # document reaches the rendering unchanged, and that a descriptor
+          # built without one renders no key rather than a null. Pure serde —
+          # the seam an agent reads a processor's config shape through.
           # Named one by one rather than by module prefix: a prefix would
           # enrol whatever that module holds next, which is how a device test
           # joins this list without anyone deciding to add it.
@@ -275,6 +288,8 @@ jobs:
             core::json_schema::port_rendering_tests::port_info_output_renders_exactly_the_declared_keys \
             core::json_schema::port_rendering_tests::port_info_output_carries_no_type_key_under_any_spelling \
             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::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 \

packages/test-fixtures/src/test_fixture_processor_configs.rs

@@ -5,72 +5,83 @@
 //! pinned at.
 
 use serde::{Deserialize, Serialize};
+use streamlib::sdk::schemars::JsonSchema;
 
 /// Compute-kernel CPU-reference fixture: buffer length and where to write the
 /// comparison result.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct ComputeKernelTestProcessorConfig {
     pub element_count: u32,
     pub output_path: String,
 }
 
 /// Concurrent-escalate fixture: how many threads contend and how long each
 /// holds the gate.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct ConcurrentEscalateTestProcessorConfig {
     pub hold_ms: u32,
     pub output_path: String,
     pub thread_count: u32,
 }
 
 /// Escalate smoke fixture: where to record that the round trip completed.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct EscalateSmokeTestProcessorConfig {
     pub output_path: String,
 }
 
 /// GPU-acquire fixture: the pixel-buffer dimensions to acquire.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct GpuAcquireTestProcessorConfig {
     pub height: u32,
     pub output_path: String,
     pub width: u32,
 }
 
 /// Graphics-kernel smoke fixture: where to record that the render completed.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct GraphicsKernelSmokeTestProcessorConfig {
     pub output_path: String,
 }
 
 /// Lifecycle-probe fixture: how many process iterations to run and where to
 /// append the per-hook markers.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct LifecycleProbeProcessorConfig {
     pub max_iterations: u32,
     pub output_path: String,
 }
 
 /// Panic-injection Continuous fixture: which lifecycle hook panics.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct PanickingContinuousLifecycleProcessorConfig {
     pub panic_at_hook: String,
 }
 
 /// Panic-injection Manual fixture: which lifecycle hook panics.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct PanickingManualLifecycleProcessorConfig {
     pub panic_at_hook: String,
 }
 
 /// Ray-tracing-kernel smoke fixture: where to record that the trace completed.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct RayTracingKernelSmokeTestProcessorConfig {
     pub output_path: String,
 }
 
 /// Attribute-macro config-emit fixture: one scalar field to round-trip.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct TestConfiguredProcessorConfig {
     pub threshold: f32,
 }

runtime/streamlib-api-server/src/api_server_config.rs

@@ -5,9 +5,11 @@
 //! pinned at.
 
 use serde::{Deserialize, Serialize};
+use streamlib::sdk::schemars::JsonSchema;
 
 /// Configuration for the runtime API server.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[schemars(crate = "streamlib::sdk::schemars")]
 pub struct ApiServerConfig {
     /// Host address to bind to.
     pub host: String,

runtime/streamlib-api-server/src/handlers.rs

@@ -631,6 +631,9 @@ mod router_surface_and_auth_gate_tests {
         Request, StatusCode,
         header::{AUTHORIZATION, CONTENT_TYPE},
     };
+    use streamlib::sdk::descriptors::{
+        ProcessorClassImportPath, ProcessorClassShortName, ProcessorDescriptor,
+    };
     use streamlib::sdk::runtime::BoxFuture;
     use tower::ServiceExt;
 
@@ -776,6 +779,53 @@ mod router_surface_and_auth_gate_tests {
         }
     }
 
+    /// `/api/registry` reads the process-global registry, so this registers a
+    /// probe under a path no other test names and asserts on that path alone.
+    /// There is no teardown: a registration is for the life of the process, and
+    /// the registry refuses a second one of the same path.
+    ///
+    /// `/api/registry` is where an agent learns which keys a processor's
+    /// config takes, so it serves the descriptor's schema document itself —
+    /// each field's type, its description and its default — rather than a
+    /// name the agent would have to look up somewhere the node does not serve.
+    #[tokio::test]
+    async fn the_registry_serves_a_registered_processors_config_schema_document() {
+        let config_schema = serde_json::json!({
+            "type": "object",
+            "properties": {
+                "width": { "type": "integer", "description": "Frame width in pixels.", "default": 1280 },
+                "height": { "type": "integer", "description": "Frame height in pixels.", "default": 720 },
+            },
+            "required": [],
+        });
+        let class_import_path = "streamlib_api_server::registry_rendering_probe::TestPatternProbe";
+        PROCESSOR_REGISTRY
+            .register_descriptor_only(
+                ProcessorDescriptor::new(
+                    ProcessorClassShortName::new("TestPatternProbe").unwrap(),
+                    ProcessorClassImportPath::new(class_import_path).unwrap(),
+                    "a registry-rendering probe",
+                )
+                .with_config_schema(config_schema.clone()),
+            )
+            .expect("the probe's path is registered by this test alone");
+
+        let request = Request::builder()
+            .method("GET")
+            .uri("/api/registry")
+            .body(Body::empty())
+            .unwrap();
+        let served = json_body_on(auth_enabled_router(), request).await;
+
+        let probe = served["processors"]
+            .as_array()
+            .expect("a processor list")
+            .iter()
+            .find(|entry| entry["processor_class_import_path"] == class_import_path)
+            .expect("the probe the test registered");
+        assert_eq!(probe["config_schema"], config_schema);
+    }
+
     /// The spec a client is generated from and the spec the node serves must be
     /// one document. They were two hand-maintained declarations once; the copy
     /// drifted, kept publishing routes the server had dropped, and nothing went

runtime/streamlib-engine/Cargo.toml

@@ -197,6 +197,7 @@ mp4-atom = "0.15"
 serial_test = "3.2"  # Run tests sequentially to avoid global PUBSUB interference
 tempfile = "3.14"  # Temporary directories for logging and runtime tests
 criterion = { version = "0.5", features = ["html_reports"] }  # Benches for logging hot path (#447)
+trybuild = "1.0.121"
 
 [target.'cfg(target_os = "linux")'.dev-dependencies]
 # `testing::SubprocessCrashHarness` — drives a real subprocess crash against

runtime/streamlib-engine/examples/codec_roundtrip_rig.rs

@@ -75,6 +75,7 @@ mod linux_rig {
     use streamlib::sdk::media_clock::MediaClock;
     use streamlib::sdk::processors::ContinuousProcessor;
     use streamlib::sdk::rhi::{PixelBuffer, PixelFormat, PublishedPixelBufferFrameId};
+    use streamlib::sdk::schemars::JsonSchema;
     use streamlib_media_builtins::mp4_annex_b_access_unit::{
         NAL_UNIT_LENGTH_PREFIX_BYTES, annex_b_access_unit_from_length_prefixed_sample,
     };
@@ -162,7 +163,8 @@ mod linux_rig {
     }
 
     /// Configuration for [`PsnrReferenceFixtureSource`].
-    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
+    #[schemars(crate = "streamlib::sdk::schemars")]
     pub struct PsnrReferenceFixtureSourceConfig {
         /// Directory of reference PNGs, replayed in sorted filename order.
         #[serde(default = "default_fixtures_directory")]
@@ -451,7 +453,8 @@ mod linux_rig {
     ///
     /// `Default` is the empty pair every processor config owes; the rig always
     /// states both, and an unset path is refused by name at `setup()`.
-    #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+    #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
+    #[schemars(crate = "streamlib::sdk::schemars")]
     pub struct RecordedMp4TrackReplaySourceConfig {
         /// The recording to replay.
         recording_path: String,

runtime/streamlib-engine/src/core/descriptors.rs

@@ -8,9 +8,9 @@
 //! re-exports them here so every `crate::core::descriptors::*` path resolves
 //! unchanged.
 
+pub use streamlib_processor_schema::config_schema_document::ProcessorConfigJsonSchema;
 pub use streamlib_processor_schema::descriptors::{
-    CodeExamples, ConfigDescriptor, ConfigField, PortDescriptor, ProcessorDescriptor,
-    ProcessorRuntime,
+    CodeExamples, PortDescriptor, ProcessorDescriptor, ProcessorRuntime,
 };
 pub use streamlib_processor_schema::{
     AUDIO_WINDOW_CHANNELS_FOLLOWING_THE_SOURCE, AUDIO_WINDOW_DTYPE_DECLARATION_VALUES,