feat: enforce input_schema at call time (ADR-016 INVALID_INPUT leg)

- OperationSpec.input_schema was advertise-only: services/schema
  disclosed it but no dispatch entry point consulted it (the only
  enforced schema was publish_schema per-chunk on Pub ops, P-03).
- compile input_schema once at registration, same fail-closed rule as
  publish_schema/CF-003: an un-compilable schema is a registration
  error, never a silently-skipped contract. Validator cache mirrored
  on fork and in OperationRegistryBuilder like the publish validators.
- check after the ACL gate in all three dispatch entry points:
  invoke, invoke_streaming, invoke_sink (via resolve_sink_handler,
  preserving the P-08 single-source-of-truth property). Violations
  return INVALID_INPUT with the input echoed in details.
- raw-JSON-Schema semantics (permissive on unknown keys); adapters
  wanting closed-by-default keep their own hardening (alkhttp's
  CompiledInputSchema composes unchanged).
- motivated by alktty review #001 L1: the channels open-op wrapper
  hands the registry-checked input to the OpenHandler as the
  authoritative params, which requires the registry to validate it.

Verification: cargo test 596 lib (6 new: invoke/streaming/sink
enforcement, fail-closed registration, permissive-{} compile,
fork-carries-validator); --all-features 613; clippy -D warnings
clean; fmt clean; doc 0 warnings; publish --dry-run ok. alkhttp
438+16 tests pass against 0.3.1 (registry) — re-verify against the
published 0.4.0 after upload.

CHANGELOG.md

@@ -4,6 +4,45 @@ All notable changes to this crate are documented here. The format is
 based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
 this crate adheres to [Semantic Versioning](https://semver.org/).
 
+## [0.4.0] - 2026-09-05
+
+`input_schema` is now enforced at call time (the advertise-vs-enforce
+leg ADR-016 promised: `INVALID_INPUT` is the registry's
+schema-mismatch error code). Minor bump — behavioral break for any
+caller that was passing schema-violating inputs to ops declaring a
+non-trivial `input_schema` and relying on the validation being
+documentation-only.
+
+### Changed
+
+- **`OperationSpec.input_schema` is enforced by the registry on every
+  dispatch.** Until now the schema was advertise-only: `services/schema`
+  disclosed it, the wire and gateway paths disclosed it, but no dispatch
+  entry point consulted it (the only schema alkcall actually enforced
+  was `publish_schema` per-chunk on Pub ops, P-03). The schema now
+  compiles once at registration (same fail-closed rule as
+  `publish_schema`/CF-003: an un-compilable schema is a registration
+  error, never a silently-skipped contract) and is checked after the
+  ACL gate in all three dispatch entry points — `invoke`,
+  `invoke_streaming`, and `invoke_sink` (via `resolve_sink_handler`,
+  preserving the P-08 single-source-of-truth property). Violations
+  return `CallError::invalid_input("input failed input_schema
+  validation")` with the input echoed in `details`. Raw-JSON-Schema
+  semantics (permissive on unknown keys) — adapters that want
+  closed-by-default enforcement keep doing their own hardening, as
+  alkhttp's `CompiledInputSchema` already does; the two checks compose.
+  Built-in ops are unaffected: alkcall's own specs declare the empty
+  object schema (accepts anything) and `services/schema` already
+  returned `INVALID_INPUT` for a missing `name` before its handler ran.
+  Motivated by alktty's code review #001 (L1): the channels open-op
+  wrapper hands the registry-checked `input` to the `OpenHandler` as
+  the authoritative params, which requires the registry to actually
+  validate it.
+- `OperationRegistry` gained an `input_validator` cache (compiled at
+  registration, mirrored on `fork` and in `OperationRegistryBuilder`
+  like the publish validators); `input_validator()` is `pub(crate)` —
+  the public surface is unchanged.
+
 ## [0.3.1] - 2026-09-04
 
 Bug-fix release: `services/list-peers` can now list peer-announced

Cargo.lock

@@ -27,7 +27,7 @@ dependencies = [
 
 [[package]]
 name = "alkcall"
-version = "0.3.1"
+version = "0.4.0"
 dependencies = [
  "async-trait",
  "bytes",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "alkcall"
-version = "0.3.1"
+version = "0.4.0"
 edition = "2021"
 rust-version = "1.85"
 license = "MIT OR Apache-2.0"

src/registry/registration.rs

@@ -109,6 +109,7 @@ impl HandlerRegistration {
 pub struct OperationRegistry {
     operations: RwLock<HashMap<String, HandlerRegistration>>,
     validators: RwLock<HashMap<String, Option<jsonschema::Validator>>>,
+    input_validators: RwLock<HashMap<String, Option<jsonschema::Validator>>>,
 }
 
 impl OperationRegistry {
@@ -116,23 +117,27 @@ impl OperationRegistry {
         Self {
             operations: RwLock::new(HashMap::new()),
             validators: RwLock::new(HashMap::new()),
+            input_validators: RwLock::new(HashMap::new()),
         }
     }
 
     /// Fork the registry: a deep copy of every registration (handlers
-    /// and provenance included) plus the cached publish-schema
-    /// validators (review 004 F-02/F-03). The fork is an independent
-    /// registry — later registrations on the original do not appear on
-    /// the fork and vice versa. It is the per-session composition seam
-    /// (ADR-047 §4 amendment): fork the base registry, register the
-    /// per-session ops (generic channel ops, openables, bootstrap ops)
-    /// against the fork, and dispatch the session over the fork.
+    /// and provenance included) plus the cached publish-schema and
+    /// input-schema validators (review 004 F-02/F-03). The fork is an
+    /// independent registry — later registrations on the original do
+    /// not appear on the fork and vice versa. It is the per-session
+    /// composition seam (ADR-047 §4 amendment): fork the base
+    /// registry, register the per-session ops (generic channel ops,
+    /// openables, bootstrap ops) against the fork, and dispatch the
+    /// session over the fork.
     pub fn fork(&self) -> OperationRegistry {
         let operations = self.operations.read().clone();
         let validators = self.validators.read().clone();
+        let input_validators = self.input_validators.read().clone();
         OperationRegistry {
             operations: RwLock::new(operations),
             validators: RwLock::new(validators),
+            input_validators: RwLock::new(input_validators),
         }
     }
 
@@ -163,6 +168,36 @@ impl OperationRegistry {
         Ok(())
     }
 
+    /// Compile a spec's `input_schema` for call-time enforcement
+    /// (ADR-016: `INVALID_INPUT` is the registry's schema-mismatch
+    /// error code; the advertise-vs-enforce leg every protocol crate
+    /// leans on — the channels open-op path passes the registry-checked
+    /// `input` to its `OpenHandler` as the authoritative params).
+    /// Un-compilable schemas are rejected at registration time — the
+    /// same fail-closed rule as `publish_schema` (CF-003): registering
+    /// an un-validatable input contract must never produce an operation
+    /// that silently skips validation. A compiled validator is cached
+    /// so the dispatch path does zero per-request work; `None` means
+    /// "no schema declared" (the empty-object schema alkcall's own ops
+    /// declare accepts anything, but it is still compiled so the
+    /// dispatch path never re-parses it).
+    fn validate_and_cache_input_schema(
+        input_validators: &mut HashMap<String, Option<jsonschema::Validator>>,
+        spec: &OperationSpec,
+    ) -> Result<(), String> {
+        let compiled = match jsonschema::options().build(&spec.input_schema) {
+            Ok(v) => Some(v),
+            Err(e) => {
+                return Err(format!(
+                    "input_schema for operation {:?} failed to compile: {e}",
+                    spec.name
+                ));
+            }
+        };
+        input_validators.insert(spec.name.clone(), compiled);
+        Ok(())
+    }
+
     pub fn register(&self, registration: HandlerRegistration) -> Result<(), String> {
         let expected = match registration.spec.op_type {
             OperationType::Query | OperationType::Mutation => "Once",
@@ -181,6 +216,10 @@ impl OperationRegistry {
             ));
         }
         Self::validate_and_cache_publish_schema(&mut self.validators.write(), &registration.spec)?;
+        Self::validate_and_cache_input_schema(
+            &mut self.input_validators.write(),
+            &registration.spec,
+        )?;
         self.operations
             .write()
             .insert(registration.spec.name.clone(), registration);
@@ -199,6 +238,41 @@ impl OperationRegistry {
         self.validators.read().get(name).cloned().flatten()
     }
 
+    /// The compiled call-time validator for an op's `input_schema`
+    /// (ADR-016: `INVALID_INPUT` is the registry's schema-mismatch
+    /// error). Compiled at registration time (CF-003 fail-closed — an
+    /// un-compilable schema is a registration error, never a silent
+    /// skip). Every declared schema compiles to `Some`; the `Option`
+    /// exists for map-miss on an unknown op name.
+    pub(crate) fn input_validator(&self, name: &str) -> Option<jsonschema::Validator> {
+        self.input_validators.read().get(name).cloned().flatten()
+    }
+
+    /// Validate `input` against the op's compiled `input_schema`
+    /// (ADR-016). Returns the `INVALID_INPUT` envelope on violation.
+    /// Called by all three dispatch entry points (`invoke`,
+    /// `invoke_streaming`, `invoke_sink` via `resolve_sink_handler`)
+    /// after the ACL check and before the handler — the registry-level
+    /// gate every protocol crate's dispatch funnels through.
+    fn check_input_schema(
+        &self,
+        registration: &HandlerRegistration,
+        input: &Value,
+        request_id: &str,
+    ) -> Result<(), ResponseEnvelope> {
+        let Some(validator) = self.input_validator(&registration.spec.name) else {
+            return Ok(());
+        };
+        if validator.is_valid(input) {
+            return Ok(());
+        }
+        Err(ResponseEnvelope::error(
+            request_id.to_string(),
+            CallError::invalid_input("input failed input_schema validation")
+                .with_details(serde_json::json!({ "input": input })),
+        ))
+    }
+
     pub fn list_operations(&self) -> Vec<OperationSpec> {
         self.operations
             .read()
@@ -248,6 +322,10 @@ impl OperationRegistry {
             return ResponseEnvelope::forbidden(request_id, message);
         }
 
+        if let Err(envelope) = self.check_input_schema(&registration, &input, &request_id) {
+            return envelope;
+        }
+
         match &registration.handler {
             HandlerKind::Once(handler) => {
                 let handler = Arc::clone(handler);
@@ -316,6 +394,10 @@ impl OperationRegistry {
             }));
         }
 
+        if let Err(envelope) = self.check_input_schema(&registration, &input, &request_id) {
+            return Box::pin(stream::once(async move { envelope }));
+        }
+
         let streaming_handler = match &registration.handler {
             HandlerKind::Stream(h) => Arc::clone(h),
             HandlerKind::Once(_) => {
@@ -395,6 +477,8 @@ impl OperationRegistry {
             return Err(ResponseEnvelope::forbidden(request_id, message));
         }
 
+        self.check_input_schema(&registration, input, &request_id)?;
+
         match &registration.handler {
             HandlerKind::Sink(h) => Ok(Arc::clone(h)),
             HandlerKind::Once(_) => Err(ResponseEnvelope::error(
@@ -441,6 +525,7 @@ impl Default for OperationRegistry {
 pub struct OperationRegistryBuilder {
     operations: HashMap<String, HandlerRegistration>,
     validators: HashMap<String, Option<jsonschema::Validator>>,
+    input_validators: HashMap<String, Option<jsonschema::Validator>>,
 }
 
 impl OperationRegistryBuilder {
@@ -448,6 +533,7 @@ impl OperationRegistryBuilder {
         Self {
             operations: HashMap::new(),
             validators: HashMap::new(),
+            input_validators: HashMap::new(),
         }
     }
 
@@ -460,6 +546,7 @@ impl OperationRegistryBuilder {
         Self {
             operations: registry.operations.read().clone(),
             validators: registry.validators.read().clone(),
+            input_validators: registry.input_validators.read().clone(),
         }
     }
 
@@ -468,6 +555,10 @@ impl OperationRegistryBuilder {
             &mut self.validators,
             &registration.spec,
         )?;
+        OperationRegistry::validate_and_cache_input_schema(
+            &mut self.input_validators,
+            &registration.spec,
+        )?;
         let name = registration.spec.name.clone();
         self.operations.insert(name, registration);
         Ok(self)
@@ -477,6 +568,7 @@ impl OperationRegistryBuilder {
         OperationRegistry {
             operations: RwLock::new(self.operations),
             validators: RwLock::new(self.validators),
+            input_validators: RwLock::new(self.input_validators),
         }
     }
 
@@ -2631,4 +2723,222 @@ mod tests {
         let response = registry.invoke("late/op", serde_json::json!({}), ctx).await;
         assert!(response.result.is_ok());
     }
+
+    // --- input_schema call-time enforcement (ADR-016 INVALID_INPUT leg) ----
+
+    use futures::stream::StreamExt;
+
+    fn spec_with_input_schema(name: &str, op_type: OperationType, schema: Value) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            op_type,
+            Visibility::External,
+            schema,
+            serde_json::json!({}),
+            vec![],
+            AccessControl::default(),
+            None,
+        )
+    }
+
+    #[tokio::test]
+    async fn invoke_rejects_input_failing_input_schema() {
+        let registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec_with_input_schema(
+                    "echo",
+                    OperationType::Query,
+                    serde_json::json!({
+                        "type": "object",
+                        "properties": { "n": { "type": "integer" } },
+                        "required": ["n"]
+                    }),
+                ),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+
+        let ctx = root_context("req-schema-1", None, None, false, ScopedPeerEnv::empty());
+        let response = registry.invoke("echo", serde_json::json!({}), ctx).await;
+        let err = response
+            .result
+            .expect_err("missing required field must be rejected");
+        assert_eq!(err.code, "INVALID_INPUT");
+
+        let ctx = root_context("req-schema-2", None, None, false, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke("echo", serde_json::json!({"n": "not-an-int"}), ctx)
+            .await;
+        assert!(response.result.is_err(), "type mismatch must be rejected");
+
+        let ctx = root_context("req-schema-3", None, None, false, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke("echo", serde_json::json!({"n": 1}), ctx)
+            .await;
+        assert_eq!(response.result.unwrap(), serde_json::json!({"n": 1}));
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_rejects_input_failing_input_schema() {
+        let registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec_with_input_schema(
+                    "sub",
+                    OperationType::Sub,
+                    serde_json::json!({
+                        "type": "object",
+                        "properties": { "channel": { "type": "string" } },
+                        "required": ["channel"]
+                    }),
+                ),
+                HandlerKind::Stream(make_streaming_handler(|_input, ctx| {
+                    Box::pin(futures::stream::once(async move {
+                        ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
+                    })) as Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+
+        let ctx = root_context("req-sub-1", None, None, false, ScopedPeerEnv::empty());
+        let mut responses: Vec<_> = registry
+            .invoke_streaming("sub", serde_json::json!({}), ctx)
+            .collect()
+            .await;
+        assert_eq!(responses.len(), 1);
+        let response = responses.pop().expect("one envelope");
+        let err = response
+            .result
+            .expect_err("schema violation must be rejected");
+        assert_eq!(err.code, "INVALID_INPUT");
+
+        let ctx = root_context("req-sub-2", None, None, false, ScopedPeerEnv::empty());
+        let responses: Vec<_> = registry
+            .invoke_streaming("sub", serde_json::json!({"channel": "c"}), ctx)
+            .collect()
+            .await;
+        assert_eq!(responses.len(), 1);
+        assert!(responses[0].result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_rejects_input_failing_input_schema() {
+        let registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec_with_input_schema(
+                    "pub",
+                    OperationType::Pub,
+                    serde_json::json!({
+                        "type": "object",
+                        "properties": { "n": { "type": "integer" } },
+                        "required": ["n"]
+                    }),
+                ),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+
+        let (chunk_tx, chunk_rx) = futures::channel::mpsc::channel::<Result<Value, CallError>>(8);
+        drop(chunk_tx);
+        let publish_stream: PublishStream = Box::pin(chunk_rx);
+        let ctx = root_context("req-pub-1", None, None, false, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke_sink("pub", serde_json::json!({}), publish_stream, ctx)
+            .await;
+        let err = response
+            .result
+            .expect_err("schema violation must be rejected");
+        assert_eq!(err.code, "INVALID_INPUT");
+    }
+
+    #[test]
+    fn uncompilable_input_schema_is_rejected_at_registration() {
+        let registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            spec_with_input_schema(
+                "broken",
+                OperationType::Query,
+                serde_json::json!({
+                    "required": "n"
+                }),
+            ),
+            HandlerKind::Once(echo_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        assert!(
+            result.is_err(),
+            "un-compilable input_schema must fail registration"
+        );
+        assert!(
+            result.unwrap_err().contains("input_schema"),
+            "error must name the failing schema"
+        );
+    }
+
+    #[test]
+    fn empty_object_input_schema_compiles_and_accepts_anything() {
+        let registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec_with_input_schema("echo", OperationType::Query, serde_json::json!({})),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        assert!(
+            registry.input_validator("echo").is_some(),
+            "a declared schema compiles even when permissive"
+        );
+    }
+
+    #[tokio::test]
+    async fn fork_carries_input_validator_and_enforces() {
+        let registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec_with_input_schema(
+                    "echo",
+                    OperationType::Query,
+                    serde_json::json!({
+                        "type": "object",
+                        "properties": { "n": { "type": "integer" } },
+                        "required": ["n"]
+                    }),
+                ),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+
+        let fork = registry.fork();
+        let ctx = root_context("req-fork-schema", None, None, false, ScopedPeerEnv::empty());
+        let response = fork.invoke("echo", serde_json::json!({}), ctx).await;
+        let err = response
+            .result
+            .expect_err("the fork must carry the compiled input validator");
+        assert_eq!(err.code, "INVALID_INPUT");
+    }
 }