diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7271c8735..dfcf71da0 100644 --- a/.github/workflows/test.yml +++ b/.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 \ diff --git a/Cargo.lock b/Cargo.lock index 7858c6b96..a67c1b480 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4384,6 +4384,7 @@ dependencies = [ "tonic-build", "tracing", "tracing-subscriber", + "trybuild", "utoipa", "uuid", "winit", @@ -4550,6 +4551,12 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca" +[[package]] +name = "target-tuple" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "876fef147edbcbddc8ac5cbbba92c7b86519e314e86638596c09673b2ed01e7f" + [[package]] name = "tatolab-vulkanalia" version = "0.35.0" @@ -4593,6 +4600,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "termcolor" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755" +dependencies = [ + "winapi-util", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -4749,6 +4765,21 @@ dependencies = [ "winnow 0.7.15", ] +[[package]] +name = "toml" +version = "1.0.6+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "399b1124a3c9e16766831c6bba21e50192572cdd98706ea114f9502509686ffc" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned 1.0.4", + "toml_datetime 1.0.0+spec-1.1.0", + "toml_parser", + "toml_writer", + "winnow 0.7.15", +] + [[package]] name = "toml_datetime" version = "0.6.11" @@ -4943,6 +4974,21 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "trybuild" +version = "1.0.121" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0cabaa10be1917331a313866bd94526343e03c77bcf69144b62b072ad35d47c" +dependencies = [ + "glob", + "serde", + "serde_derive", + "serde_json", + "target-tuple", + "termcolor", + "toml 1.0.6+spec-1.1.0", +] + [[package]] name = "ttf-parser" version = "0.25.1" diff --git a/packages/test-fixtures/src/test_fixture_processor_configs.rs b/packages/test-fixtures/src/test_fixture_processor_configs.rs index f84537166..4973f6746 100644 --- a/packages/test-fixtures/src/test_fixture_processor_configs.rs +++ b/packages/test-fixtures/src/test_fixture_processor_configs.rs @@ -5,10 +5,12 @@ //! 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, @@ -16,7 +18,8 @@ pub struct ComputeKernelTestProcessorConfig { /// 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, @@ -24,13 +27,15 @@ pub struct ConcurrentEscalateTestProcessorConfig { } /// 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, @@ -38,39 +43,45 @@ pub struct GpuAcquireTestProcessorConfig { } /// 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, } diff --git a/runtime/streamlib-api-server/src/api_server_config.rs b/runtime/streamlib-api-server/src/api_server_config.rs index 0c943c0ae..1f5c74dd5 100644 --- a/runtime/streamlib-api-server/src/api_server_config.rs +++ b/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, diff --git a/runtime/streamlib-api-server/src/handlers.rs b/runtime/streamlib-api-server/src/handlers.rs index 5dff63644..f8d371224 100644 --- a/runtime/streamlib-api-server/src/handlers.rs +++ b/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 diff --git a/runtime/streamlib-engine/Cargo.toml b/runtime/streamlib-engine/Cargo.toml index 1152a78b3..953b00711 100644 --- a/runtime/streamlib-engine/Cargo.toml +++ b/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 diff --git a/runtime/streamlib-engine/examples/codec_roundtrip_rig.rs b/runtime/streamlib-engine/examples/codec_roundtrip_rig.rs index cd12798b9..2de36f002 100644 --- a/runtime/streamlib-engine/examples/codec_roundtrip_rig.rs +++ b/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, diff --git a/runtime/streamlib-engine/src/core/descriptors.rs b/runtime/streamlib-engine/src/core/descriptors.rs index f1561f864..ed7ba337b 100644 --- a/runtime/streamlib-engine/src/core/descriptors.rs +++ b/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, diff --git a/runtime/streamlib-engine/src/core/json_schema.rs b/runtime/streamlib-engine/src/core/json_schema.rs index 56f98ccad..fecf96fcc 100644 --- a/runtime/streamlib-engine/src/core/json_schema.rs +++ b/runtime/streamlib-engine/src/core/json_schema.rs @@ -204,9 +204,10 @@ pub struct ProcessorDescriptorOutput { /// Entrypoint for non-Rust runtimes. #[serde(default, skip_serializing_if = "Option::is_none")] pub entrypoint: Option, - /// Reference to config schema. + /// The config type's JSON Schema, as JSON Schema draft 2020-12 — what an + /// agent reads to learn which keys this processor's config takes. #[serde(default, skip_serializing_if = "Option::is_none")] - pub config_schema: Option, + pub config_schema: Option, /// Input port descriptors. pub inputs: Vec, /// Output port descriptors. @@ -215,20 +216,6 @@ pub struct ProcessorDescriptorOutput { pub examples: CodeExamplesOutput, } -/// A configuration field for a processor. -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, utoipa::ToSchema)] -pub struct ConfigFieldOutput { - /// Field name. - pub name: String, - /// Field type as string (e.g., "String", "u32", "Option"). - #[serde(rename = "type")] - pub field_type: String, - /// Whether the field is required. - pub required: bool, - /// Human-readable description. - pub description: String, -} - /// Descriptor for a processor port. #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, utoipa::ToSchema)] pub struct PortDescriptorOutput { @@ -435,17 +422,6 @@ impl From<&crate::core::ProcessorRuntime> for ProcessorRuntimeOutput { } } -impl From<&crate::core::ConfigField> for ConfigFieldOutput { - fn from(field: &crate::core::ConfigField) -> Self { - Self { - name: field.name.clone(), - field_type: field.field_type.clone(), - required: field.required, - description: field.description.clone(), - } - } -} - impl From<&crate::core::PortDescriptor> for PortDescriptorOutput { fn from(port: &crate::core::PortDescriptor) -> Self { Self { @@ -861,3 +837,53 @@ mod capability_extension_rendering_tests { ); } } + +#[cfg(test)] +mod config_schema_rendering_tests { + use super::*; + use crate::core::descriptors::{ + ProcessorClassImportPath, ProcessorClassShortName, ProcessorDescriptor, + }; + + fn descriptor_carrying(config_schema: Option) -> ProcessorDescriptor { + let descriptor = ProcessorDescriptor::new( + ProcessorClassShortName::new("TestPatternSource").unwrap(), + ProcessorClassImportPath::new("streamlib_media_builtins::test_pattern_source").unwrap(), + "a probe", + ); + match config_schema { + Some(document) => descriptor.with_config_schema(document), + None => descriptor, + } + } + + /// The control plane serves the descriptor's document, not a summary of + /// it: a field's type, its description and its default all survive the + /// hop, because the MCP catalog reads the same rendering. + #[test] + fn a_registered_descriptors_config_schema_reaches_the_rendering_unchanged() { + let document = 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 }, + }, + }); + let rendered = serde_json::to_value(ProcessorDescriptorOutput::from(&descriptor_carrying( + Some(document.clone()), + ))) + .unwrap(); + assert_eq!(rendered["config_schema"], document); + } + + /// A descriptor built without one — every Python descriptor today — + /// renders no key at all rather than a null an agent would have to read + /// as "no config". + #[test] + fn a_descriptor_carrying_no_config_schema_renders_no_key_rather_than_a_null() { + let rendered = + serde_json::to_value(ProcessorDescriptorOutput::from(&descriptor_carrying(None))) + .unwrap(); + assert!(rendered.get("config_schema").is_none(), "{rendered}"); + } +} diff --git a/runtime/streamlib-engine/src/core/processors/empty_config.rs b/runtime/streamlib-engine/src/core/processors/empty_config.rs new file mode 100644 index 000000000..df99e9f5d --- /dev/null +++ b/runtime/streamlib-engine/src/core/processors/empty_config.rs @@ -0,0 +1,82 @@ +// Copyright (c) 2025 Jonathan Fontanez +// SPDX-License-Identifier: BUSL-1.1 + +//! The config type of a processor that declares none. + +/// The config type of a processor that declares none. +/// +/// It publishes an empty-object schema and takes nothing: a configuration +/// carrying keys is refused rather than discarded, because a processor that +/// declares no config cannot act on one and silently dropping it hides a +/// wiring mistake. It serializes back as an empty named map so it round-trips +/// as a bag. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct EmptyConfig; + +impl serde::Serialize for EmptyConfig { + fn serialize( + &self, + serializer: S, + ) -> std::result::Result { + serde::ser::SerializeMap::end(serializer.serialize_map(Some(0))?) + } +} + +impl<'de> serde::Deserialize<'de> for EmptyConfig { + fn deserialize>( + deserializer: D, + ) -> std::result::Result { + deserializer.deserialize_any(EmptyConfigVisitor) + } +} + +/// Accepts an empty named map and the legacy `nil`; refuses anything a +/// processor declaring no config could not have meant. +struct EmptyConfigVisitor; + +impl<'de> serde::de::Visitor<'de> for EmptyConfigVisitor { + type Value = EmptyConfig; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("an empty configuration") + } + + fn visit_map>( + self, + mut named_map: A, + ) -> std::result::Result { + match named_map.next_key::()? { + Some(refused_key) => Err(serde::de::Error::custom(format!( + "this processor declares no config and takes none, \ + so `{refused_key}` has nowhere to go" + ))), + None => Ok(EmptyConfig), + } + } + + fn visit_unit(self) -> std::result::Result { + Ok(EmptyConfig) + } +} + +impl schemars::JsonSchema for EmptyConfig { + fn schema_name() -> String { + "EmptyConfig".to_string() + } + + fn json_schema(_: &mut schemars::r#gen::SchemaGenerator) -> schemars::schema::Schema { + schemars::schema::SchemaObject { + instance_type: Some(schemars::schema::InstanceType::Object.into()), + metadata: Some(Box::new(schemars::schema::Metadata { + description: Some("This processor declares no configuration.".to_string()), + ..Default::default() + })), + object: Some(Box::new(schemars::schema::ObjectValidation { + additional_properties: Some(Box::new(schemars::schema::Schema::Bool(false))), + ..Default::default() + })), + ..Default::default() + } + .into() + } +} diff --git a/runtime/streamlib-engine/src/core/processors/mod.rs b/runtime/streamlib-engine/src/core/processors/mod.rs index 6153e49ce..ba64abf1b 100644 --- a/runtime/streamlib-engine/src/core/processors/mod.rs +++ b/runtime/streamlib-engine/src/core/processors/mod.rs @@ -8,6 +8,7 @@ pub mod traits; #[doc(hidden)] pub mod __generated_private; +mod empty_config; mod processor_instance_factory; mod processor_spec; // Re-export graph types — `ProcessorState` and `ProcessorStateComponent` @@ -26,38 +27,12 @@ pub use __generated_private::{ DynGeneratedProcessor, GeneratedProcessor, OutOfProcessLinkWiringEnvelope, }; +pub use empty_config::EmptyConfig; pub use processor_instance_factory::{ DynamicProcessorConstructorFn, PROCESSOR_REGISTRY, ProcessorInstance, ProcessorInstanceFactory, }; pub use processor_spec::ProcessorSpec; -/// Empty config type for processors that don't need configuration. -/// -/// Config-as-bag delivers config as a named map, so `EmptyConfig` must -/// tolerate any wire shape: an empty map `{}`, a legacy `nil`, or a -/// populated map (whose fields it discards). It serializes back as an -/// empty named map so it round-trips as a bag. -#[derive(Debug, Clone, Default, PartialEq)] -pub struct EmptyConfig; - -impl serde::Serialize for EmptyConfig { - fn serialize( - &self, - serializer: S, - ) -> std::result::Result { - serde::ser::SerializeMap::end(serializer.serialize_map(Some(0))?) - } -} - -impl<'de> serde::Deserialize<'de> for EmptyConfig { - fn deserialize>( - deserializer: D, - ) -> std::result::Result { - deserializer.deserialize_ignored_any(serde::de::IgnoredAny)?; - Ok(EmptyConfig) - } -} - // Audio processors are not here: capture and playback ship as engine // built-ins in `streamlib-media-builtins` (`microphone_source.rs`, // `speaker_sink.rs`). diff --git a/runtime/streamlib-engine/src/lib.rs b/runtime/streamlib-engine/src/lib.rs index 27dc0cf35..1f36c3d43 100644 --- a/runtime/streamlib-engine/src/lib.rs +++ b/runtime/streamlib-engine/src/lib.rs @@ -17,6 +17,7 @@ extern crate self as streamlib; // Re-export crossbeam_channel for macro-generated code pub use crossbeam_channel; pub use inventory; +pub use schemars; pub use serde_json; pub mod core; @@ -31,11 +32,10 @@ pub mod logging { pub use crate::core::logging::*; } -// Re-export attribute macros for processor syntax: -// - #[streamlib::processor(execution = …, …)] - execution + ports in code; a -// processor's identity is the import path of its type, captured by the macro -// - #[derive(ConfigDescriptor)] - Config field metadata derive macro -pub use streamlib_macros::{ConfigDescriptor, processor}; +// Re-export the attribute macro for processor syntax: +// #[streamlib::processor(execution = …, …)] - execution + ports in code; a +// processor's identity is the import path of its type, captured by the macro +pub use streamlib_macros::processor; /// The `#[processor]` attribute accepts no authored identity, in any spelling. /// @@ -302,9 +302,10 @@ pub mod sdk { pub use crate::iceoryx2; pub use crate::inventory; pub use crate::logging; + pub use crate::schemars; pub use crate::serde_json; - pub use streamlib_macros::{ConfigDescriptor, processor}; + pub use streamlib_macros::processor; pub mod permissions { pub use crate::{ diff --git a/runtime/streamlib-engine/tests/attribute_macro_test.rs b/runtime/streamlib-engine/tests/attribute_macro_test.rs index db265e35e..a326e45e6 100644 --- a/runtime/streamlib-engine/tests/attribute_macro_test.rs +++ b/runtime/streamlib-engine/tests/attribute_macro_test.rs @@ -9,6 +9,8 @@ //! intentionally does not register the processor in the global //! `PROCESSOR_REGISTRY`. +use serde::{Deserialize, Serialize}; +use streamlib::sdk::schemars::JsonSchema; use streamlib_engine::core::GeneratedProcessor; use streamlib_engine::core::{EmptyConfig, Result, RuntimeContextFullAccess}; @@ -86,21 +88,101 @@ fn test_processor_instantiation() { } #[test] -fn empty_config_is_a_tolerant_bag() { - // config-as-bag: a no-config processor's `EmptyConfig` deserializes from - // any named map, discarding unknown / forward-compat keys, and serializes - // back as an empty named map. Mentally revert the custom EmptyConfig serde - // impls and this fails (a unit struct rejects a map). - let from_populated: EmptyConfig = - serde_json::from_value(serde_json::json!({ "leftover": 1, "future": true })).unwrap(); - let processor = TestProcessor::Processor::from_config(from_populated).unwrap(); - assert_eq!(processor.name(), "TestProcessor"); +fn a_processor_declaring_no_config_takes_none_and_says_which_key_had_nowhere_to_go() { + let refusal = serde_json::from_value::(serde_json::json!({ "leftover": 1 })) + .expect_err("a populated configuration must be refused, not discarded"); + assert!( + refusal.to_string().contains("leftover"), + "the refusal must name the key with nowhere to go: {refusal}" + ); + // The two shapes a no-config processor is legitimately added with: the + // empty named map `ProcessorSpec` serializes, and the legacy `nil`. let from_empty: EmptyConfig = serde_json::from_value(serde_json::json!({})).unwrap(); assert_eq!( serde_json::to_value(from_empty).unwrap(), serde_json::json!({}) ); + serde_json::from_value::(serde_json::Value::Null) + .expect("a nil configuration is still no configuration"); + + let processor = TestProcessor::Processor::from_config(EmptyConfig).unwrap(); + assert_eq!(processor.name(), "TestProcessor"); +} + +#[test] +fn a_processor_declaring_no_config_publishes_an_empty_object_schema() { + let descriptor = TestProcessor::Processor::descriptor().expect("a descriptor"); + let config_schema = descriptor + .config_schema + .expect("every macro-emitted descriptor carries a document"); + assert_eq!(config_schema["type"], "object"); + assert_eq!(config_schema["additionalProperties"], false); + assert!(config_schema.get("properties").is_none()); +} + +/// Configuration for [`ConfiguredProbeProcessor`] — the shape the emitted +/// document is read back from. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] +pub struct ConfiguredProbeProcessorConfig { + /// Frame width in pixels. + #[serde(default = "default_probe_width")] + pub width: u32, + /// Where the recording is written. + pub path: String, +} + +fn default_probe_width() -> u32 { + 1280 +} + +impl Default for ConfiguredProbeProcessorConfig { + fn default() -> Self { + Self { + width: default_probe_width(), + path: String::new(), + } + } +} + +#[streamlib::sdk::processor( + execution = manual, + config = crate::ConfiguredProbeProcessorConfig, + output("video_out"), +)] +pub struct ConfiguredProbeProcessor; + +impl streamlib_engine::ManualProcessor for ConfiguredProbeProcessor::Processor { + fn start(&mut self, _ctx: &RuntimeContextFullAccess<'_>) -> Result<()> { + Ok(()) + } +} + +#[test] +fn the_descriptor_carries_the_config_types_schema_rather_than_its_name() { + let descriptor = ConfiguredProbeProcessor::Processor::descriptor().expect("a descriptor"); + let config_schema = descriptor + .config_schema + .expect("a configured processor carries its config type's document"); + + assert_eq!(config_schema["properties"]["width"]["type"], "integer"); + assert_eq!( + config_schema["properties"]["width"]["description"], + "Frame width in pixels." + ); + assert_eq!(config_schema["properties"]["width"]["default"], 1280); + assert_eq!( + config_schema["required"], + serde_json::json!(["path"]), + "a field serde declares no default for is the only required one" + ); + // The retired id grammar named the type; the document describes it. + assert!(config_schema.get("$schema").is_none()); + assert_ne!( + config_schema, + serde_json::json!("ConfiguredProbeProcessorConfig") + ); } #[test] diff --git a/runtime/streamlib-engine/tests/compile_fail/config_without_json_schema.rs b/runtime/streamlib-engine/tests/compile_fail/config_without_json_schema.rs new file mode 100644 index 000000000..c7f1831ae --- /dev/null +++ b/runtime/streamlib-engine/tests/compile_fail/config_without_json_schema.rs @@ -0,0 +1,25 @@ +// Copyright (c) 2025 Jonathan Fontanez +// SPDX-License-Identifier: BUSL-1.1 + +use serde::{Deserialize, Serialize}; +use streamlib::sdk::context::RuntimeContextFullAccess; +use streamlib::sdk::error::Result; + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ProbeConfigWithoutTheDerive { + pub width: u32, +} + +#[streamlib::sdk::processor( + execution = manual, + config = crate::ProbeConfigWithoutTheDerive, +)] +pub struct ProbeProcessor; + +impl streamlib::sdk::processors::ManualProcessor for ProbeProcessor::Processor { + fn start(&mut self, _ctx: &RuntimeContextFullAccess<'_>) -> Result<()> { + Ok(()) + } +} + +fn main() {} diff --git a/runtime/streamlib-engine/tests/compile_fail/config_without_json_schema.stderr b/runtime/streamlib-engine/tests/compile_fail/config_without_json_schema.stderr new file mode 100644 index 000000000..cda7c6e63 --- /dev/null +++ b/runtime/streamlib-engine/tests/compile_fail/config_without_json_schema.stderr @@ -0,0 +1,24 @@ +error[E0277]: `ProbeConfigWithoutTheDerive` is a processor `config =` type but does not derive `JsonSchema` + --> tests/compile_fail/config_without_json_schema.rs:15:14 + | +15 | config = crate::ProbeConfigWithoutTheDerive, + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ unsatisfied trait bound + | +help: the trait `JsonSchema` is not implemented for `ProbeConfigWithoutTheDerive` + --> tests/compile_fail/config_without_json_schema.rs:9:1 + | + 9 | pub struct ProbeConfigWithoutTheDerive { + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + = note: add `#[derive(streamlib::sdk::schemars::JsonSchema)]` and `#[schemars(crate = "streamlib::sdk::schemars")]` to `ProbeConfigWithoutTheDerive` + = note: the SDK re-exports `schemars` at `streamlib::sdk::schemars`, so the crate needs no new dependency + = help: the following other types implement trait `JsonSchema`: + &'a T + &'a mut T + () + (T0, T1) + (T0, T1, T2) + (T0, T1, T2, T3) + (T0, T1, T2, T3, T4) + (T0, T1, T2, T3, T4, T5) + and $N others + = note: required for `ProbeConfigWithoutTheDerive` to implement `ProcessorConfigJsonSchema` diff --git a/runtime/streamlib-engine/tests/compile_fail_config_without_json_schema.rs b/runtime/streamlib-engine/tests/compile_fail_config_without_json_schema.rs new file mode 100644 index 000000000..20f73b505 --- /dev/null +++ b/runtime/streamlib-engine/tests/compile_fail_config_without_json_schema.rs @@ -0,0 +1,19 @@ +// Copyright (c) 2025 Jonathan Fontanez +// SPDX-License-Identifier: BUSL-1.1 + +//! The refusal an author meets when a `config =` type does not derive +//! `JsonSchema`. +//! +//! The descriptor carries the config type's schema, so the derive is a +//! requirement rather than an option. Without a compile-fail case nothing runs +//! the compiler over a config type that lacks it, and the note that names the +//! fix could rot into a bare trait-bound error with every other test green. +//! +//! Refresh the expected output with `TRYBUILD=overwrite cargo test -p +//! streamlib-engine --test compile_fail_config_without_json_schema` after a +//! compiler upgrade reflows a diagnostic. + +#[test] +fn a_config_type_without_the_derive_is_refused_with_the_derive_and_its_path() { + trybuild::TestCases::new().compile_fail("tests/compile_fail/*.rs"); +} diff --git a/runtime/streamlib-media-builtins/src/audio_window_to_encoded_packet_encoder.rs b/runtime/streamlib-media-builtins/src/audio_window_to_encoded_packet_encoder.rs index 4bd667e24..4c074ca69 100644 --- a/runtime/streamlib-media-builtins/src/audio_window_to_encoded_packet_encoder.rs +++ b/runtime/streamlib-media-builtins/src/audio_window_to_encoded_packet_encoder.rs @@ -25,6 +25,7 @@ use serde::{Deserialize, Serialize}; use streamlib::sdk::error::{Error, Result}; +use streamlib::sdk::schemars::JsonSchema; use crate::audio_block::{AudioBlock, AudioSampleDtype}; use crate::encoded_audio_packet::{EncodedAudioCodec, EncodedAudioPacket}; @@ -51,7 +52,8 @@ const MULTISTREAM_PACKET_FRAMING_HEADROOM_BYTES: usize = 256; /// Which libopus tuning an [`OpusEncoderConfig`] asks for, spelled the way /// the wire spells it. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub enum OpusEncoderApplication { /// Broadcast and high fidelity: the decoded audio should be as close as /// possible to the input. The default, because a recording rung wants @@ -86,7 +88,8 @@ impl OpusEncoderApplication { /// redundancy a recording never reads, and DTX replaces silence with nothing /// — a gap the plan's own doctrine says must stay derivable from the stamps /// rather than be invented back by a decoder. -#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct OpusEncoderConfig { /// Target bitrate in bits per second. Absent, libopus picks its own from /// the sample rate and channel count. diff --git a/runtime/streamlib-media-builtins/src/camera_source.rs b/runtime/streamlib-media-builtins/src/camera_source.rs index e2d2ee731..d064d968e 100644 --- a/runtime/streamlib-media-builtins/src/camera_source.rs +++ b/runtime/streamlib-media-builtins/src/camera_source.rs @@ -29,6 +29,7 @@ use streamlib::sdk::rhi::{ PixelBuffer, PixelFormat, RhiColorConverter, SourceLayoutInfo, StorageBuffer, Texture, TextureFormat, VulkanLayout, }; +use streamlib::sdk::schemars::JsonSchema; use v4l::FourCC; use v4l::buffer::Type; @@ -53,7 +54,8 @@ const RING_SLOT_WAIT_TIMEOUT_NS: u64 = 2_000_000_000; const HOST_READBACK_WAIT_TIMEOUT_NS: u64 = 5_000_000_000; /// Configuration for [`CameraSource`]. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct CameraSourceConfig { /// V4L2 device path (`/dev/video0`). Absent: the first capture-capable /// device found. diff --git a/runtime/streamlib-media-builtins/src/display_window.rs b/runtime/streamlib-media-builtins/src/display_window.rs index 4c51845b0..43210a84d 100644 --- a/runtime/streamlib-media-builtins/src/display_window.rs +++ b/runtime/streamlib-media-builtins/src/display_window.rs @@ -30,6 +30,7 @@ use streamlib::sdk::processor_owned_window::{ SurfaceNamedForPresentationOnOwnedWindow, }; use streamlib::sdk::processors::ManualProcessor; +use streamlib::sdk::schemars::JsonSchema; use streamlib::sdk::window_event_pump::WindowRegistrationRequestFromOwningProcessor; use crate::video_frame::{ColorInfo, VideoFrame}; @@ -44,7 +45,8 @@ const DISPLAY_RENDER_THREAD_IDLE_PARK_INTERVAL: Duration = Duration::from_millis const DEGRADED_DISPLAY_DRAIN_PARK_INTERVAL: Duration = Duration::from_millis(2); /// How the frame maps onto the window, as configuration vocabulary. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] #[serde(rename_all = "snake_case")] pub enum DisplayScaling { /// Whole frame visible, black bars fill the rest. @@ -67,7 +69,8 @@ impl DisplayScaling { } /// Configuration for [`DisplayWindow`]. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct DisplayWindowConfig { /// Window title. #[serde(default = "default_title")] diff --git a/runtime/streamlib-media-builtins/src/encoded_frame_to_published_surface_decoder.rs b/runtime/streamlib-media-builtins/src/encoded_frame_to_published_surface_decoder.rs index ede6752ad..446e27726 100644 --- a/runtime/streamlib-media-builtins/src/encoded_frame_to_published_surface_decoder.rs +++ b/runtime/streamlib-media-builtins/src/encoded_frame_to_published_surface_decoder.rs @@ -37,6 +37,7 @@ use streamlib::sdk::engine::video::decode::{ }; use streamlib::sdk::error::{Error, Result}; use streamlib::sdk::rhi::{PixelBuffer, PixelFormat}; +use streamlib::sdk::schemars::JsonSchema; use crate::cumulative_count_report_threshold::CumulativeCountReportThreshold; use crate::encoded_stream_ordering::{ArrivingEncodedBagDisposition, EncodedStreamSyncPointGate}; @@ -61,7 +62,8 @@ const STREAM_RE_ENTRY_REPORT_INTERVAL: u64 = 20; /// absent, the extent is auto-detected from the stream's first SPS, which is /// what a decoder fed by an unknown producer wants. The DPB's slot count is /// the session surface's own and is not configurable here. -#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct HardwareVideoDecoderConfig { /// Upper bound on the coded width the DPB is allocated for. Absent: /// auto-detected from the first SPS. diff --git a/runtime/streamlib-media-builtins/src/encoded_video_frame.rs b/runtime/streamlib-media-builtins/src/encoded_video_frame.rs index da3adfde9..64f1cf4ba 100644 --- a/runtime/streamlib-media-builtins/src/encoded_video_frame.rs +++ b/runtime/streamlib-media-builtins/src/encoded_video_frame.rs @@ -26,10 +26,12 @@ use serde::{Deserialize, Serialize}; use crate::video_frame::ColorInfo; +use streamlib::sdk::schemars::JsonSchema; /// Elementary-stream identity of an encoded frame's bitstream, spelled the /// way the wire spells it. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub enum EncodedVideoCodec { #[serde(rename = "h264")] H264, diff --git a/runtime/streamlib-media-builtins/src/microphone_source.rs b/runtime/streamlib-media-builtins/src/microphone_source.rs index bd8c33ea3..b98d94b58 100644 --- a/runtime/streamlib-media-builtins/src/microphone_source.rs +++ b/runtime/streamlib-media-builtins/src/microphone_source.rs @@ -23,6 +23,7 @@ use streamlib::sdk::context::{ use streamlib::sdk::error::{Error, Result}; use streamlib::sdk::iceoryx2::OutputWriter; use streamlib::sdk::processors::ManualProcessor; +use streamlib::sdk::schemars::JsonSchema; use crate::audio_block::{AudioBlock, AudioSampleDtype}; use crate::captured_audio_block_hand_off_ring::{ @@ -64,7 +65,8 @@ const AUDIO_OUTPUT_PORT: &str = "audio"; const PUBLISH_THREAD_EXIT_GRACE: Duration = Duration::from_secs(2); /// Configuration for [`MicrophoneSource`]. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct MicrophoneSourceConfig { /// Backend-named capture device. Absent: the backend's default device. /// A name the backend cannot open raises rather than landing on a diff --git a/runtime/streamlib-media-builtins/src/mp4_sink.rs b/runtime/streamlib-media-builtins/src/mp4_sink.rs index 9990d65c6..41072afe5 100644 --- a/runtime/streamlib-media-builtins/src/mp4_sink.rs +++ b/runtime/streamlib-media-builtins/src/mp4_sink.rs @@ -19,6 +19,7 @@ use serde::{Deserialize, Serialize}; use streamlib::sdk::context::{RuntimeContextFullAccess, RuntimeContextLimitedAccess}; use streamlib::sdk::error::{Error, Result}; use streamlib::sdk::processors::ReactiveProcessor; +use streamlib::sdk::schemars::JsonSchema; use crate::mp4_fragmented_file_writer::Mp4FragmentedFileWriter; @@ -29,7 +30,8 @@ pub const MP4_SINK_PROCESSOR_NAME: &str = "Mp4Sink"; const SILENT_LINK_REPORT_INTERVAL: Duration = Duration::from_secs(1); /// Where the recording is written. -#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct Mp4SinkConfig { /// The file to write, created or truncated at `setup()`. /// diff --git a/runtime/streamlib-media-builtins/src/published_surface_to_encoded_frame_encoder.rs b/runtime/streamlib-media-builtins/src/published_surface_to_encoded_frame_encoder.rs index 7fd79228f..2725d93c9 100644 --- a/runtime/streamlib-media-builtins/src/published_surface_to_encoded_frame_encoder.rs +++ b/runtime/streamlib-media-builtins/src/published_surface_to_encoded_frame_encoder.rs @@ -24,6 +24,7 @@ use serde::{Deserialize, Serialize}; use streamlib::sdk::context::{GpuContextLimitedAccess, RuntimeContextFullAccess}; use streamlib::sdk::engine::video::{EncodePacket, Preset, SimpleEncoder, SimpleEncoderConfig}; use streamlib::sdk::error::{Error, Result}; +use streamlib::sdk::schemars::JsonSchema; use crate::encoded_stream_ordering::EncodedStreamOrderingPairCounter; use crate::encoded_video_frame::EncodedVideoFrame; @@ -44,7 +45,8 @@ const ENCODE_PROGRESS_LOG_INTERVAL_FRAMES: u64 = 300; /// optional: dimensions and rate track the upstream frames, and the knobs /// below are the guardrail set the session surface accepts. Both codecs take /// exactly these, because the session surface takes exactly these. -#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct HardwareVideoEncoderConfig { /// Expected frame width — a guardrail, not a resize: a mismatching /// frame wins with a warning. diff --git a/runtime/streamlib-media-builtins/src/speaker_sink.rs b/runtime/streamlib-media-builtins/src/speaker_sink.rs index fc72c9a58..cb5a92bea 100644 --- a/runtime/streamlib-media-builtins/src/speaker_sink.rs +++ b/runtime/streamlib-media-builtins/src/speaker_sink.rs @@ -28,6 +28,7 @@ use streamlib::sdk::context::{ use streamlib::sdk::error::{Error, Result}; use streamlib::sdk::iceoryx2::{AudioWindowContractMatchingADeviceStream, InputMailboxes}; use streamlib::sdk::processors::ManualProcessor; +use streamlib::sdk::schemars::JsonSchema; use crate::audio_block::AudioBlock; use crate::audio_samples_awaiting_playback_ring::{ @@ -83,7 +84,8 @@ const AUDIO_INPUT_PORT: &str = "audio"; const DRAIN_THREAD_EXIT_GRACE: Duration = Duration::from_secs(2); /// Configuration for [`SpeakerSink`]. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct SpeakerSinkConfig { /// Backend-named playback device. Absent: the backend's default device. /// A name the backend cannot open raises rather than landing on a diff --git a/runtime/streamlib-media-builtins/src/test_pattern_source.rs b/runtime/streamlib-media-builtins/src/test_pattern_source.rs index e8723ac51..7dcb41fe7 100644 --- a/runtime/streamlib-media-builtins/src/test_pattern_source.rs +++ b/runtime/streamlib-media-builtins/src/test_pattern_source.rs @@ -10,12 +10,14 @@ use streamlib::sdk::error::Result; use streamlib::sdk::media_clock::MediaClock; use streamlib::sdk::processors::ContinuousProcessor; use streamlib::sdk::rhi::{PixelBuffer, PixelFormat, PublishedPixelBufferFrameId}; +use streamlib::sdk::schemars::JsonSchema; use crate::video_frame::{ColorInfo, Primaries, Range, Transfer, VideoFrame}; /// Configuration for [`TestPatternSource`]: frame size only — the pattern, /// rate, and pixel format are fixed. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct TestPatternSourceConfig { /// Frame width in pixels. #[serde(default = "default_width")] @@ -186,6 +188,7 @@ impl ContinuousProcessor for TestPatternSource::Processor { #[cfg(test)] mod tests { use super::*; + use streamlib::sdk::processors::GeneratedProcessor; #[test] fn bars_cover_the_full_width_in_order() { @@ -226,4 +229,24 @@ mod tests { let config: TestPatternSourceConfig = serde_json::from_str("{}").expect("empty config"); assert_eq!((config.width, config.height), (1280, 720)); } + + /// What an agent reads before it adds this processor. The defaults live in + /// two `#[serde(default = …)]` functions and nowhere else, so a document + /// that does not carry them is one the agent has to guess against. + #[test] + fn the_descriptor_publishes_each_config_fields_type_description_and_default() { + let descriptor = ::descriptor() + .expect("a descriptor"); + let config_schema = descriptor.config_schema.expect("a config schema"); + + for (field, default, description) in [ + ("width", 1280, "Frame width in pixels."), + ("height", 720, "Frame height in pixels."), + ] { + let rendered = &config_schema["properties"][field]; + assert_eq!(rendered["type"], "integer", "{field}"); + assert_eq!(rendered["default"], default, "{field}"); + assert_eq!(rendered["description"], description, "{field}"); + } + } } diff --git a/runtime/streamlib-media-builtins/src/virtual_camera_sink.rs b/runtime/streamlib-media-builtins/src/virtual_camera_sink.rs index 90e5f49d4..b0b25568c 100644 --- a/runtime/streamlib-media-builtins/src/virtual_camera_sink.rs +++ b/runtime/streamlib-media-builtins/src/virtual_camera_sink.rs @@ -40,6 +40,7 @@ use streamlib::sdk::engine::host_rhi::{ use streamlib::sdk::error::{Error, Result}; use streamlib::sdk::processors::ReactiveProcessor; use streamlib::sdk::rhi::{PixelFormat, RhiColorConverter, VulkanLayout}; +use streamlib::sdk::schemars::JsonSchema; use crate::cumulative_count_report_threshold::CumulativeCountReportThreshold; use crate::v4l2_color::resolved_color_to_v4l2_color; @@ -106,7 +107,8 @@ const V4L2_PIX_FMT_PRIV_MAGIC: u32 = v4l::v4l_sys::V4L2_PIX_FMT_PRIV_MAGIC; const WRITE_FAILURE_REPORT_STEP: u64 = 300; /// Which door a [`VirtualCameraSink`] takes. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] #[serde(rename_all = "snake_case")] pub enum VirtualCameraDoor { /// The loopback door when the control node is writable, else PipeWire. @@ -121,7 +123,8 @@ pub enum VirtualCameraDoor { } /// Configuration for [`VirtualCameraSink`]. -#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct VirtualCameraSinkConfig { /// The camera's name in every picker. Absent: `StreamLib Camera` plus a /// short id that is unique per instance and app and stable across runs. diff --git a/sdk/streamlib-macros/src/codegen.rs b/sdk/streamlib-macros/src/codegen.rs index ad086a69c..48a41fca9 100644 --- a/sdk/streamlib-macros/src/codegen.rs +++ b/sdk/streamlib-macros/src/codegen.rs @@ -21,15 +21,14 @@ use syn::{ItemStruct, Path}; /// /// `config_type_path` is the Rust type path for the processor's typed `Config` /// alias, taken verbatim from the attribute's `config = `; `None` binds -/// the tolerant [`EmptyConfig`]. `config_field_name` is the generated struct -/// field (present iff `config_type_path` is `Some`). `config_schema_id` is the -/// descriptor-metadata id string emitted into `with_config_schema(...)`. +/// `EmptyConfig`, whose schema is the empty object a processor declaring no +/// config publishes. `config_field_name` is the generated struct field +/// (present iff `config_type_path` is `Some`). pub fn generate_from_processor_schema( item: &ItemStruct, schema: &ProcessorSchema, config_type_path: Option<&Path>, config_field_name: Option<&str>, - config_schema_id: Option<&str>, sdk_root: TokenStream, ) -> TokenStream { let module_name = &item.ident; @@ -54,7 +53,6 @@ pub fn generate_from_processor_schema( &config_type, &config_field_name, &custom_fields, - config_schema_id, ); let processor_class_import_path_accessor = quote! { @@ -315,7 +313,6 @@ fn generate_processor_impl_from_schema( config_type: &TokenStream, config_field_name: &Option, custom_fields: &[CustomField], - config_schema_id: Option<&str>, ) -> TokenStream { use streamlib_processor_schema::ProcessorSchemaExecution; @@ -391,7 +388,7 @@ fn generate_processor_impl_from_schema( let from_config_body = generate_from_config_from_schema(schema, config_field_name, custom_fields); - let descriptor_impl = generate_descriptor_from_schema(schema, description, config_schema_id); + let descriptor_impl = generate_descriptor_from_schema(schema, description, config_type); let iceoryx2_accessors = generate_iceoryx2_accessors_from_schema(schema); let update_config = config_field_name.as_ref().map(|name| { @@ -564,14 +561,13 @@ fn generate_from_config_from_schema( /// Generate descriptor method from schema. /// -/// `config_schema_id` is the descriptor-metadata id string emitted into -/// `with_config_schema(...)`, declared (or synthesized from the config type) -/// by the `#[processor(...)]` attribute. `None` when the processor declares -/// no config. +/// `config_type` is the processor's `Config` type — the declared `config =` +/// path, or `EmptyConfig` where none was declared. Its JSON Schema is what the +/// descriptor carries. fn generate_descriptor_from_schema( schema: &ProcessorSchema, description: &str, - config_schema_id: Option<&str>, + config_type: &TokenStream, ) -> TokenStream { let repository = "https://github.com/tatolab/streamlib"; @@ -621,13 +617,15 @@ fn generate_descriptor_from_schema( }) .collect(); - // Config schema reference (descriptor metadata), declared or synthesized - // by the attribute. Emitted verbatim into `with_config_schema(...)`. - let config_schema = config_schema_id.map(|schema_ref| { - quote! { - .with_config_schema(#schema_ref) - } - }); + // The config type's JSON Schema. Reached through the bound-carrying + // trait rather than `schemars` directly, so a config type missing the + // derive fails on a diagnostic that names the fix. + let config_schema = quote! { + .with_config_schema( + <#config_type as __streamlib_sdk::descriptors::ProcessorConfigJsonSchema> + ::processor_config_schema_document() + ) + }; // Declarative scheduling intent. Absent → `Normal` priority. The OS // thread name is derived by the compiler from the processor type + node @@ -1288,14 +1286,7 @@ mod processor_struct_emit_tests { } fn expand_probe_processor(item: &ItemStruct) -> TokenStream { - generate_from_processor_schema( - item, - &minimal_schema(), - None, - None, - None, - quote! { streamlib }, - ) + generate_from_processor_schema(item, &minimal_schema(), None, None, quote! { streamlib }) } fn struct_with_cfg_attr_on_a_field() -> ItemStruct { @@ -1442,10 +1433,16 @@ mod processor_struct_emit_tests { } fn rendered_descriptor() -> String { + rendered_descriptor_for_config_type("e! { + __streamlib_sdk::processors::EmptyConfig + }) + } + + fn rendered_descriptor_for_config_type(config_type: &TokenStream) -> String { render_token_stream_without_whitespace(generate_descriptor_from_schema( &minimal_schema(), "a probe", - None, + config_type, )) } @@ -1457,7 +1454,6 @@ mod processor_struct_emit_tests { "e! { __streamlib_sdk::processors::EmptyConfig }, &None, &[], - None, )) } @@ -1484,6 +1480,39 @@ mod processor_struct_emit_tests { ); } + /// The descriptor's config slot carries the config type's schema, derived + /// through the bound-carrying trait — never a name, and never `schemars` + /// reached directly, which would fail on a bound the author has no path to. + #[test] + fn the_descriptor_derives_its_config_schema_from_the_declared_config_type() { + let rendered = rendered_descriptor_for_config_type("e! { + crate::camera_source::CameraSourceConfig + }); + assert!( + rendered.contains( + "with_config_schema(::\ + processor_config_schema_document())" + ), + "the descriptor must derive the declared config type's schema — got: {rendered}" + ); + } + + /// A processor declaring no config still publishes a document: the empty + /// config's, which is an object with no properties. + #[test] + fn a_processor_declaring_no_config_still_carries_the_empty_configs_schema() { + let rendered = rendered_descriptor(); + assert!( + rendered.contains( + "with_config_schema(<__streamlib_sdk::processors::EmptyConfigas\ + __streamlib_sdk::descriptors::ProcessorConfigJsonSchema>::\ + processor_config_schema_document())" + ), + "a no-config processor must still carry a schema — got: {rendered}" + ); + } + /// `std::any::type_name`'s output format is documented as unspecified and /// free to change between compiler versions. Keying a registry on it means /// a toolchain bump silently renames every processor, with nothing failing diff --git a/sdk/streamlib-macros/src/config_descriptor.rs b/sdk/streamlib-macros/src/config_descriptor.rs deleted file mode 100644 index 986fb8411..000000000 --- a/sdk/streamlib-macros/src/config_descriptor.rs +++ /dev/null @@ -1,228 +0,0 @@ -// Copyright (c) 2025 Jonathan Fontanez -// SPDX-License-Identifier: BUSL-1.1 - -//! Derive macro for ConfigDescriptor trait. - -use proc_macro2::TokenStream; -use quote::quote; -use syn::{Data, DeriveInput, Error, Fields, GenericArgument, PathArguments, Result, Type}; - -/// Extract doc comments from attributes as a single description string. -fn extract_doc_comments(attrs: &[syn::Attribute]) -> String { - attrs - .iter() - .filter_map(|attr| { - if attr.path().is_ident("doc") { - if let syn::Meta::NameValue(nv) = &attr.meta { - if let syn::Expr::Lit(syn::ExprLit { - lit: syn::Lit::Str(lit_str), - .. - }) = &nv.value - { - return Some(lit_str.value().trim().to_string()); - } - } - } - None - }) - .collect::>() - .join(" ") -} - -/// Convert a Rust type to a string representation for the config field. -fn type_to_string(ty: &Type) -> String { - match ty { - Type::Path(type_path) => { - let segment = type_path.path.segments.last(); - if let Some(seg) = segment { - let ident = seg.ident.to_string(); - - // Handle Option - if ident == "Option" { - if let PathArguments::AngleBracketed(args) = &seg.arguments { - if let Some(GenericArgument::Type(inner_ty)) = args.args.first() { - return format!("Option<{}>", type_to_string(inner_ty)); - } - } - } - - // Handle Vec - if ident == "Vec" { - if let PathArguments::AngleBracketed(args) = &seg.arguments { - if let Some(GenericArgument::Type(inner_ty)) = args.args.first() { - return format!("Vec<{}>", type_to_string(inner_ty)); - } - } - } - - // Handle other generic types - if let PathArguments::AngleBracketed(args) = &seg.arguments { - let inner_types: Vec = args - .args - .iter() - .filter_map(|arg| { - if let GenericArgument::Type(inner_ty) = arg { - Some(type_to_string(inner_ty)) - } else { - None - } - }) - .collect(); - if !inner_types.is_empty() { - return format!("{}<{}>", ident, inner_types.join(", ")); - } - } - - ident - } else { - "unknown".to_string() - } - } - Type::Array(arr) => { - let elem = type_to_string(&arr.elem); - // Try to extract the array length - if let syn::Expr::Lit(syn::ExprLit { - lit: syn::Lit::Int(lit_int), - .. - }) = &arr.len - { - format!("[{}; {}]", elem, lit_int) - } else { - format!("[{}; N]", elem) - } - } - Type::Tuple(tuple) => { - let elems: Vec = tuple.elems.iter().map(type_to_string).collect(); - format!("({})", elems.join(", ")) - } - Type::Reference(reference) => { - let inner = type_to_string(&reference.elem); - if reference.mutability.is_some() { - format!("&mut {}", inner) - } else { - format!("&{}", inner) - } - } - _ => "unknown".to_string(), - } -} - -/// Check if a type is Option -fn is_option_type(ty: &Type) -> bool { - if let Type::Path(type_path) = ty { - if let Some(seg) = type_path.path.segments.last() { - return seg.ident == "Option"; - } - } - false -} - -/// Generate the ConfigDescriptor derive implementation. -pub fn derive_config_descriptor(input: DeriveInput) -> Result { - let struct_name = &input.ident; - - // Extract fields from struct - let fields = match &input.data { - Data::Struct(data_struct) => match &data_struct.fields { - Fields::Named(fields_named) => &fields_named.named, - Fields::Unit => { - // Unit struct - no fields - return Ok(quote! { - impl ::streamlib::sdk::descriptors::ConfigDescriptor for #struct_name { - fn config_fields() -> ::std::vec::Vec<::streamlib::sdk::descriptors::ConfigField> { - ::std::vec::Vec::new() - } - } - }); - } - _ => { - return Err(Error::new( - input.ident.span(), - "ConfigDescriptor can only be derived for structs with named fields or unit structs", - )); - } - }, - _ => { - return Err(Error::new( - input.ident.span(), - "ConfigDescriptor can only be derived for structs", - )); - } - }; - - // Generate field descriptors - let field_descriptors: Vec = fields - .iter() - .filter_map(|field| { - let field_name = field.ident.as_ref()?.to_string(); - let field_type = type_to_string(&field.ty); - let required = !is_option_type(&field.ty); - let description = extract_doc_comments(&field.attrs); - - Some(quote! { - ::streamlib::sdk::descriptors::ConfigField { - name: #field_name.to_string(), - field_type: #field_type.to_string(), - required: #required, - description: #description.to_string(), - } - }) - }) - .collect(); - - // Generate the implementation - let expanded = quote! { - impl ::streamlib::sdk::descriptors::ConfigDescriptor for #struct_name { - fn config_fields() -> ::std::vec::Vec<::streamlib::sdk::descriptors::ConfigField> { - vec![ - #(#field_descriptors),* - ] - } - } - }; - - Ok(expanded) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_type_to_string_primitives() { - // This is a unit test for the type_to_string function - // We can't easily test with actual Type objects, but we verify the logic is sound - assert_eq!(type_to_string(&syn::parse_quote!(u32)), "u32"); - assert_eq!(type_to_string(&syn::parse_quote!(String)), "String"); - assert_eq!(type_to_string(&syn::parse_quote!(bool)), "bool"); - } - - #[test] - fn test_type_to_string_option() { - assert_eq!( - type_to_string(&syn::parse_quote!(Option)), - "Option" - ); - assert_eq!( - type_to_string(&syn::parse_quote!(Option)), - "Option" - ); - } - - #[test] - fn test_type_to_string_vec() { - assert_eq!(type_to_string(&syn::parse_quote!(Vec)), "Vec"); - } - - #[test] - fn test_type_to_string_array() { - assert_eq!(type_to_string(&syn::parse_quote!([f32; 4])), "[f32; 4]"); - } - - #[test] - fn test_is_option_type() { - assert!(is_option_type(&syn::parse_quote!(Option))); - assert!(!is_option_type(&syn::parse_quote!(String))); - assert!(!is_option_type(&syn::parse_quote!(Vec))); - } -} diff --git a/sdk/streamlib-macros/src/grammar.rs b/sdk/streamlib-macros/src/grammar.rs index 0f8b32f78..902ecc201 100644 --- a/sdk/streamlib-macros/src/grammar.rs +++ b/sdk/streamlib-macros/src/grammar.rs @@ -12,7 +12,7 @@ //! execution = manual, // reactive | manual | continuous | continuous(interval_ms = 10) //! scheduling = high, // realtime | high | normal (default: normal) //! unsafe_send, // flag — emit `unsafe impl Send` -//! config = crate::CameraConfig, // Rust type path for the typed Config alias +//! config = crate::CameraConfig, // typed Config alias; must derive `JsonSchema` //! input("video_in", delivery_profile = "newest"), //! output("video"), //! )] @@ -77,7 +77,6 @@ pub struct ParsedProcessorAttr { pub unsafe_send: bool, pub config_type: Option, pub config_field_name: String, - pub config_schema_id: Option, pub inputs: Vec, pub outputs: Vec, } @@ -153,7 +152,6 @@ fn parse_body(input: ParseStream<'_>, struct_name: &str) -> syn::Result = None; let mut config_field_name: Option = None; - let mut config_schema_id: Option = None; let mut inputs: Vec = Vec::new(); let mut outputs: Vec = Vec::new(); @@ -200,14 +198,6 @@ fn parse_body(input: ParseStream<'_>, struct_name: &str) -> syn::Result { - input.parse::()?; - // Descriptor metadata only — accepts both the new-shape - // `@org/pkg/Type@version` and legacy reverse-DNS - // `.config@` id grammars verbatim. - let lit: LitStr = input.parse()?; - config_schema_id = Some(lit.value()); - } "type" => { return Err(syn::Error::new(key.span(), class_path_rule())); } @@ -241,15 +231,6 @@ fn parse_body(input: ParseStream<'_>, struct_name: &str) -> syn::Result, struct_name: &str) -> syn::Result TokenStream { &schema, parsed.config_type.as_ref(), config_field_name.as_deref(), - parsed.config_schema_id.as_deref(), sdk_root(), ); @@ -96,39 +94,3 @@ fn sdk_root() -> proc_macro2::TokenStream { // In-engine macro use: `extern crate self as streamlib` resolves this. quote! { ::streamlib::sdk } } - -/// Derive macro for ConfigDescriptor trait. -/// -/// Generates a `ConfigDescriptor` implementation for config structs, -/// enabling automatic config field metadata extraction for processor descriptors. -/// -/// # Field Handling -/// -/// - `Option` fields are marked as `required: false` -/// - All other fields are marked as `required: true` -/// - Doc comments on fields become the `description` -/// -/// # Example -/// -/// ```ignore -/// use streamlib::sdk::ConfigDescriptor; -/// -/// #[derive(ConfigDescriptor)] -/// pub struct CameraConfig { -/// /// Camera device identifier -/// pub device_id: Option, -/// /// Target width in pixels -/// pub width: u32, -/// /// Target height in pixels -/// pub height: u32, -/// } -/// ``` -#[proc_macro_derive(ConfigDescriptor)] -pub fn derive_config_descriptor(input: TokenStream) -> TokenStream { - let input = parse_macro_input!(input as DeriveInput); - - match config_descriptor::derive_config_descriptor(input) { - Ok(tokens) => tokens.into(), - Err(err) => err.to_compile_error().into(), - } -} diff --git a/sdk/streamlib-processor-schema/src/config_schema_document.rs b/sdk/streamlib-processor-schema/src/config_schema_document.rs new file mode 100644 index 000000000..768f31b5a --- /dev/null +++ b/sdk/streamlib-processor-schema/src/config_schema_document.rs @@ -0,0 +1,248 @@ +// Copyright (c) 2025 Jonathan Fontanez +// SPDX-License-Identifier: BUSL-1.1 + +//! The JSON Schema a processor descriptor carries for its config type. +//! +//! One dialect leaves this seam: JSON Schema draft 2020-12 with no `$schema` +//! key. `schemars` 0.8 emits draft-07, and the conversion is three things: +//! the meta-schema key and the pointer prefix, which the generator's own +//! settings decide; the root keyword `definitions`, which `RootSchema` +//! hard-codes; and a tuple field's positional item schemas, which draft-07 +//! spells as an array-valued `items`. + +use schemars::schema::{SchemaObject, SingleOrVec}; +use schemars::visit::{Visitor, visit_schema_object}; +use serde_json::Value; + +/// Where a `$ref` points in a 2020-12 document. +const DEFINITION_POINTER_PREFIX: &str = "#/$defs/"; + +/// Marks a type usable as a processor's `config =` type, and derives its +/// schema document. +/// +/// Carries the `JsonSchema` bound so that a config type without the derive +/// fails on this trait, whose note names the fix, rather than on a bare +/// `schemars` bound the author has no path to. +#[diagnostic::on_unimplemented( + message = "`{Self}` is a processor `config =` type but does not derive `JsonSchema`", + note = "add `#[derive(streamlib::sdk::schemars::JsonSchema)]` and `#[schemars(crate = \"streamlib::sdk::schemars\")]` to `{Self}`", + note = "the SDK re-exports `schemars` at `streamlib::sdk::schemars`, so the crate needs no new dependency" +)] +pub trait ProcessorConfigJsonSchema { + /// This config type's schema, as JSON Schema draft 2020-12. + fn processor_config_schema_document() -> Value; +} + +impl ProcessorConfigJsonSchema for T { + fn processor_config_schema_document() -> Value { + let root_schema = schemars::r#gen::SchemaSettings::draft07() + .with(|settings| { + settings.meta_schema = None; + settings.definitions_path = DEFINITION_POINTER_PREFIX.to_string(); + }) + .with_visitor(TupleItemsRewrittenAsPrefixItems) + .into_generator() + .into_root_schema_for::(); + + let mut document = + serde_json::to_value(root_schema).expect("a schemars root schema always serializes"); + rename_the_root_definitions_keyword_to_defs(&mut document); + document + } +} + +/// `RootSchema` serializes its definitions map under the draft-07 keyword +/// whatever the generator's pointer prefix says, so the root key is renamed +/// here to match the `#/$defs/` the references already carry. +fn rename_the_root_definitions_keyword_to_defs(document: &mut Value) { + if let Some(root_object) = document.as_object_mut() + && let Some(definitions) = root_object.remove("definitions") + { + root_object.insert("$defs".to_string(), definitions); + } +} + +/// Rewrites a tuple's positional item schemas into the keyword 2020-12 reads +/// them under. +/// +/// Draft-07 says positional schemas with an array-valued `items` and bounds +/// the rest with `additionalItems`; 2020-12 says `prefixItems` and lets +/// `items` mean the rest. A single-schema `items` means the same thing in both +/// and is left alone. Typed rather than a walk over the serialized document, +/// so a config type whose own `default` or `enum` data happens to hold a key +/// named `items` is never touched. +#[derive(Debug, Clone)] +struct TupleItemsRewrittenAsPrefixItems; + +impl Visitor for TupleItemsRewrittenAsPrefixItems { + fn visit_schema_object(&mut self, schema: &mut SchemaObject) { + visit_schema_object(self, schema); + + let positional_item_schemas = match schema.array.as_deref_mut() { + Some(array) if matches!(array.items, Some(SingleOrVec::Vec(_))) => { + let Some(SingleOrVec::Vec(positional_item_schemas)) = array.items.take() else { + return; + }; + array.items = array.additional_items.take().map(SingleOrVec::Single); + positional_item_schemas + } + _ => return, + }; + + schema.extensions.insert( + "prefixItems".to_string(), + Value::Array( + positional_item_schemas + .into_iter() + .map(|item_schema| { + serde_json::to_value(item_schema).expect("a schema always serializes") + }) + .collect(), + ), + ); + } +} + +/// The single-schema `items` the visitor left untouched, for the tests below. +#[cfg(test)] +fn single_schema_items(schema: &SchemaObject) -> Option<&schemars::schema::Schema> { + match schema.array.as_deref()?.items.as_ref()? { + SingleOrVec::Single(item_schema) => Some(item_schema), + SingleOrVec::Vec(_) => None, + } +} + +#[cfg(test)] +mod config_schema_document_tests { + use super::*; + use schemars::JsonSchema; + use schemars::schema::Schema; + use serde::{Deserialize, Serialize}; + + /// The shape a served document is checked against: doc comments become + /// descriptions, serde defaults become defaults, and a field without one + /// is required. + #[derive(Serialize, Deserialize, JsonSchema)] + struct FixtureSourceConfig { + /// Frame width in pixels. + #[serde(default = "default_width")] + width: u32, + /// Where the recording is written. + path: String, + /// How the frame maps onto the window. + #[serde(default)] + scaling: FixtureScaling, + /// Left, top, right, bottom. + crop: (u32, u32, u32, u32), + /// Every device to open. + device_ids: Vec, + } + + fn default_width() -> u32 { + 1280 + } + + #[derive(Serialize, Deserialize, JsonSchema, Default)] + #[serde(rename_all = "snake_case")] + enum FixtureScaling { + #[default] + Fit, + Stretch, + } + + #[test] + fn a_config_types_document_carries_each_fields_type_description_and_default() { + let document = FixtureSourceConfig::processor_config_schema_document(); + let width = &document["properties"]["width"]; + assert_eq!(width["type"], "integer"); + assert_eq!(width["description"], "Frame width in pixels."); + assert_eq!(width["default"], 1280); + } + + #[test] + fn a_field_serde_declares_no_default_for_is_required_and_one_it_does_is_not() { + let document = FixtureSourceConfig::processor_config_schema_document(); + let required = document["required"].as_array().expect("a required list"); + assert!(required.contains(&Value::String("path".to_string()))); + assert!(!required.contains(&Value::String("width".to_string()))); + assert!(!required.contains(&Value::String("scaling".to_string()))); + } + + #[test] + fn the_document_is_2020_12_with_no_schema_key_and_no_definitions_keyword() { + let document = FixtureSourceConfig::processor_config_schema_document(); + assert!(document.get("$schema").is_none()); + assert!(document.get("definitions").is_none()); + assert!(document["$defs"]["FixtureScaling"].is_object()); + } + + #[test] + fn a_nested_types_reference_points_into_defs_rather_than_definitions() { + let document = FixtureSourceConfig::processor_config_schema_document(); + let rendered = serde_json::to_string(&document).expect("the document serializes"); + assert!( + !rendered.contains("#/definitions/"), + "no draft-07 reference may survive: {rendered}" + ); + assert!( + rendered.contains("#/$defs/FixtureScaling"), + "the nested enum's reference must point into $defs: {rendered}" + ); + } + + /// Draft-07 spells a tuple's positional schemas as an array-valued + /// `items`, which 2020-12 reads as a schema for every element and refuses + /// as an array. A document labelled 2020-12 that carries the draft-07 + /// spelling is one a validator reads wrong with nothing to say so. + #[test] + fn a_tuple_fields_positional_schemas_are_named_prefix_items() { + let document = FixtureSourceConfig::processor_config_schema_document(); + let crop = &document["properties"]["crop"]; + assert_eq!(crop["type"], "array"); + assert_eq!( + crop["prefixItems"].as_array().map(Vec::len), + Some(4), + "a four-tuple carries four positional schemas: {crop}" + ); + assert!( + crop.get("items").is_none(), + "an unbounded tail was never declared, so nothing bounds it: {crop}" + ); + assert!(crop.get("additionalItems").is_none(), "{crop}"); + } + + /// The single-schema spelling means the same thing in both drafts, so a + /// plain sequence field must come through unchanged. + #[test] + fn a_sequence_fields_element_schema_is_left_as_items() { + let document = FixtureSourceConfig::processor_config_schema_document(); + let device_ids = &document["properties"]["device_ids"]; + assert_eq!(device_ids["type"], "array"); + assert_eq!(device_ids["items"]["type"], "string"); + assert!(device_ids.get("prefixItems").is_none(), "{device_ids}"); + } + + /// A tuple's draft-07 `additionalItems` is what 2020-12 calls `items`. + #[test] + fn a_bounded_tuple_tail_becomes_the_items_keyword() { + let mut schema = SchemaObject { + array: Some(Box::new(schemars::schema::ArrayValidation { + items: Some(SingleOrVec::Vec(vec![Schema::Bool(true)])), + additional_items: Some(Box::new(Schema::Bool(false))), + ..Default::default() + })), + ..Default::default() + }; + TupleItemsRewrittenAsPrefixItems.visit_schema_object(&mut schema); + + assert_eq!( + single_schema_items(&schema), + Some(&Schema::Bool(false)), + "the draft-07 tail schema becomes 2020-12's `items`" + ); + assert_eq!( + schema.extensions.get("prefixItems"), + Some(&serde_json::json!([true])) + ); + } +} diff --git a/sdk/streamlib-processor-schema/src/descriptors.rs b/sdk/streamlib-processor-schema/src/descriptors.rs index eb8948d5d..f428d522e 100644 --- a/sdk/streamlib-processor-schema/src/descriptors.rs +++ b/sdk/streamlib-processor-schema/src/descriptors.rs @@ -86,45 +86,6 @@ pub struct CodeExamples { pub typescript: String, } -/// A configuration field for a processor. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigField { - pub name: String, - #[serde(rename = "type")] - pub field_type: String, - pub required: bool, - pub description: String, -} - -impl ConfigField { - pub fn new( - name: impl Into, - field_type: impl Into, - required: bool, - description: impl Into, - ) -> Self { - Self { - name: name.into(), - field_type: field_type.into(), - required, - description: description.into(), - } - } -} - -/// Trait for config structs to provide field metadata for descriptors. -pub trait ConfigDescriptor { - /// Returns the list of config fields with their types and descriptions. - fn config_fields() -> Vec; -} - -/// Default implementation for unit type (no config). -impl ConfigDescriptor for () { - fn config_fields() -> Vec { - Vec::new() - } -} - /// Describes a processor with its ports and configuration. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ProcessorDescriptor { @@ -148,9 +109,13 @@ pub struct ProcessorDescriptor { /// Entrypoint for non-Rust runtimes (e.g., "src.blur:BlurProcessor"). #[serde(default)] pub entrypoint: Option, - /// Reference to config schema (e.g., "com.example.blur.config@1.0.0"). + /// The config type's JSON Schema, as JSON Schema draft 2020-12. + /// + /// `None` only on a descriptor built by hand without one — every + /// `#[processor]`-emitted descriptor carries a document, an empty-object + /// one where the processor declares no config. #[serde(default)] - pub config_schema: Option, + pub config_schema: Option, /// Declarative scheduling intent declared in the `#[processor]` attribute. /// Read at thread-spawn time; defaults to `Normal` priority. #[serde(default)] @@ -196,8 +161,8 @@ impl ProcessorDescriptor { self } - pub fn with_config_schema(mut self, schema: impl Into) -> Self { - self.config_schema = Some(schema.into()); + pub fn with_config_schema(mut self, config_schema: serde_json::Value) -> Self { + self.config_schema = Some(config_schema); self } diff --git a/sdk/streamlib-processor-schema/src/lib.rs b/sdk/streamlib-processor-schema/src/lib.rs index 34745a9f9..f0af7d8b3 100644 --- a/sdk/streamlib-processor-schema/src/lib.rs +++ b/sdk/streamlib-processor-schema/src/lib.rs @@ -8,6 +8,7 @@ mod process_execution; mod thread_priority; pub mod audio_window_contract; +pub mod config_schema_document; pub mod descriptors; pub mod error; pub mod processor_class_import_path; @@ -25,6 +26,7 @@ pub use audio_window_contract::{ AudioWindowContract, AudioWindowContractDeclaredValues, refuse_audio_window_beside_a_skipping_delivery_profile, render_declaration_values, }; +pub use config_schema_document::ProcessorConfigJsonSchema; pub use error::{SchemaError, SchemaResult}; pub use processor_class_import_path::ProcessorClassImportPath; pub use processor_class_short_name::ProcessorClassShortName; diff --git a/sdk/streamlib-python-wheel/src/python_test_harness_endpoints.rs b/sdk/streamlib-python-wheel/src/python_test_harness_endpoints.rs index 4cd9817ee..f5c9e8d95 100644 --- a/sdk/streamlib-python-wheel/src/python_test_harness_endpoints.rs +++ b/sdk/streamlib-python-wheel/src/python_test_harness_endpoints.rs @@ -26,12 +26,14 @@ use pyo3::prelude::*; use serde::{Deserialize, Serialize}; use streamlib::sdk::error::Result; use streamlib::sdk::processors::{ContinuousProcessor, ReactiveProcessor}; +use streamlib::sdk::schemars::JsonSchema; use crate::python_bag_conversion::{decode_msgpack_to_python_object, encode_bag_to_msgpack}; use crate::python_logging::monotonic_clock_now_ns; /// Which channel an endpoint reads from or writes to. -#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)] +#[schemars(crate = "streamlib::sdk::schemars")] pub struct TestHarnessChannelConfig { /// The name this endpoint's queue is registered under. #[serde(default)] diff --git a/sdk/streamlib-sdk/src/lib.rs b/sdk/streamlib-sdk/src/lib.rs index 5eaeb150d..95a24237d 100644 --- a/sdk/streamlib-sdk/src/lib.rs +++ b/sdk/streamlib-sdk/src/lib.rs @@ -138,6 +138,10 @@ pub mod sdk { /// `serde_json` re-export — required by macro-emitted paths. pub use streamlib_engine::serde_json; + /// `schemars` re-export — a processor crate derives `JsonSchema` on its + /// config type through this path and adds no dependency of its own. + pub use streamlib_engine::schemars; + /// `crossbeam_channel` re-export — required by macro-emitted paths. pub use streamlib_engine::crossbeam_channel; @@ -149,9 +153,6 @@ pub mod sdk { /// which the macro captures at its expansion site. pub use streamlib_engine::processor; - /// `#[derive(ConfigDescriptor)]` derive macro. - pub use streamlib_engine::ConfigDescriptor; - // ---- Permission helpers ---- pub mod permissions { diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 4355c98c9..d44842814 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -281,6 +281,18 @@ fn run_local_ci_gates(workspace_root: &Path) -> Result<()> { "attribute_macro_test", ], ), + ( + "the missing-JsonSchema-derive refusal (compile-fail)", + "cargo", + &[ + "test", + "--locked", + "-p", + "streamlib-engine", + "--test", + "compile_fail_config_without_json_schema", + ], + ), ( "media built-ins unit tests", "cargo", @@ -375,6 +387,8 @@ fn run_local_ci_gates(workspace_root: &Path) -> Result<()> { "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",