feat(http/gateway/invoke-streaming): add GatewayDispatch::invoke_streaming

Add the streaming analogue of invoke() returning BoxStream<ResponseEnvelope>.
Security invariants are identical to invoke() (internal: false,
forwarded_for: None, same capabilities/scoped_env/ACL) — shared via a
build_root_context_inner helper with a bounded flag. The streaming path
sets deadline: None (unbounded subscriptions, ADR-049 §6). Calls
OperationRegistry::invoke_streaming() (already on develop). to_mcp is
unchanged (MCP excludes Subscription, ADR-041).

Tests cover: subscription dispatch, leading-slash strip, unknown op
NOT_FOUND, internal op NOT_FOUND (not leaked), None identity FORBIDDEN,
Query op INVALID_OPERATION_TYPE, invoke() on Subscription returns
INVALID_OPERATION_TYPE (guard holds through gateway), and
build_root_context_streaming sets deadline: None while carrying the
registration bundle.

crates/alknet-call/src/client/call_client.rs

@@ -572,7 +572,7 @@ mod tests {
     use crate::protocol::connection::CallConnection;
     use crate::protocol::wire::ResponseEnvelope;
     use crate::registry::registration::{
-        make_handler, Handler, HandlerRegistration, OperationProvenance,
+        make_handler, Handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::Identity;
@@ -640,14 +640,16 @@ mod tests {
 
     fn registry_with_caps() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("pub/run"),
-            caps_inspect_handler(),
+                HandlerKind::Once(caps_inspect_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new().with_api_key("google", "pub-key".to_string()),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -709,7 +711,9 @@ mod tests {
         let client = CallClient::new(Arc::clone(&registry), Arc::new(NoopIdentityProvider));
         let conn = client.spawn_dispatch(stub_connection());
         assert_eq!(
-            conn.connection().expect("quic connection present").remote_alpn(),
+            conn.connection()
+                .expect("quic connection present")
+                .remote_alpn(),
             b"alknet/call"
         );
         std::mem::drop(conn);

crates/alknet-call/src/client/from_call.rs

@@ -19,7 +19,9 @@ use crate::client::AdapterError;
 use crate::protocol::connection::CallConnection;
 use crate::protocol::wire::ResponseEnvelope;
 use crate::registry::context::OperationContext;
-use crate::registry::registration::{Handler, HandlerRegistration, OperationProvenance};
+use crate::registry::registration::{
+    Handler, HandlerKind, HandlerRegistration, OperationProvenance, StreamingHandler,
+};
 use crate::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
 };
@@ -121,14 +123,23 @@ fn build_bundles(
             });
         }
 
-        let handler = make_forwarding_handler(
+        let kind = match spec.op_type {
+            OperationType::Subscription => HandlerKind::Stream(make_streaming_forwarding_handler(
                 Arc::new(op_summary.connection.clone()),
                 remote_name,
                 op_summary.credentials_auth_token.clone(),
-        );
+            )),
+            OperationType::Query | OperationType::Mutation => {
+                HandlerKind::Once(make_forwarding_handler(
+                    Arc::new(op_summary.connection.clone()),
+                    remote_name,
+                    op_summary.credentials_auth_token.clone(),
+                ))
+            }
+        };
         bundles.push(HandlerRegistration::new(
             spec,
-            handler,
+            kind,
             OperationProvenance::FromCall,
             None,
             None,
@@ -309,8 +320,10 @@ fn parse_access_control(v: &Value) -> AccessControl {
     }
 }
 
-/// Construct a forwarding handler for a `FromCall` leaf: on invocation, calls
+/// Construct a forwarding handler for a `FromCall` `Query`/`Mutation` leaf:
-/// the remote op via the `CallConnection` and returns its `ResponseEnvelope`.
+/// on invocation, calls the remote op via the `CallConnection` and returns
+/// its `ResponseEnvelope` (single `call_with_payload()`, `HandlerKind::Once`).
+/// `Subscription` ops use [`make_streaming_forwarding_handler`] instead.
 ///
 /// Per ADR-032 §3, the handler populates `forwarded_for` on the
 /// `call.requested` payload from the hub's `OperationContext.identity` (the
@@ -323,12 +336,6 @@ fn parse_access_control(v: &Value) -> AccessControl {
 /// If `context.identity` is `None` (the hub chose not to disclose, or has not
 /// authenticated an originator), `forwarded_for` is omitted — the spoke
 /// receives only the hub's identity.
-///
-/// For a `Subscription` op, the handler calls `subscribe` and streams until
-/// `completed`/`aborted` (the streaming path is exercised at the
-/// `CallConnection` layer; the handler here forwards the first response for
-/// query/mutation and delegates streaming to the caller via the returned
-/// envelope).
 fn make_forwarding_handler(
     connection: Arc<CallConnection>,
     remote_name: String,
@@ -357,6 +364,40 @@ fn make_forwarding_handler(
     })
 }
 
+/// Construct a streaming forwarding handler for a `FromCall` `Subscription`
+/// leaf: on invocation, calls `CallConnection::subscribe_with_payload()` and
+/// forwards the remote stream end-to-end. Each `call.responded` from the
+/// remote becomes a stream item, `call.completed` ends the stream, and
+/// `call.aborted` drops it (ADR-049 §8). No truncation, no first-value
+/// fallback.
+///
+/// `forwarded_for` is populated from `context.identity` (ADR-032 §3) and
+/// `auth_token` from the hub's own call-protocol token, exactly as the
+/// request/response forwarding handler does — both via `build_forwarded_payload`
+/// (no new payload-construction code). The `subscribe_with_payload` path
+/// registers the request in `PendingRequestMap`, so the abort cascade
+/// (ADR-016 §6) is already wired: a parent abort drops the
+/// `SubscriptionStream`, which sends `call.aborted` to the remote node.
+fn make_streaming_forwarding_handler(
+    connection: Arc<CallConnection>,
+    remote_name: String,
+    credentials_auth_token: Option<String>,
+) -> StreamingHandler {
+    use crate::registry::registration::make_streaming_handler;
+    use futures::stream::{once, StreamExt};
+    make_streaming_handler(move |input, context| {
+        let connection = Arc::clone(&connection);
+        let remote_name = remote_name.clone();
+        let auth_token = credentials_auth_token.clone();
+        once(async move {
+            let payload =
+                build_forwarded_payload(&remote_name, input, &context, auth_token.as_deref());
+            connection.subscribe_with_payload(payload).await
+        })
+        .flatten()
+    })
+}
+
 /// Build the `call.requested` payload for a forwarded call, populating
 /// `forwarded_for` from the hub's `OperationContext.identity` (ADR-032 §3).
 /// `forwarded_for` is omitted when `context.identity` is `None` (the hub
@@ -389,7 +430,7 @@ fn build_forwarded_payload(
 mod tests {
     use super::*;
     use crate::protocol::connection::CallConnection;
-    use crate::registry::registration::make_handler;
+    use crate::registry::registration::{make_handler, make_streaming_handler};
     use crate::registry::spec::OperationType;
     use alknet_core::auth::Identity;
     use alknet_core::types::{Capabilities, MockConnection};
@@ -549,7 +590,7 @@ mod tests {
         );
         let reg = HandlerRegistration::new(
             spec,
-            handler,
+            HandlerKind::Once(handler),
             OperationProvenance::FromCall,
             None,
             None,
@@ -722,6 +763,15 @@ mod tests {
         }
     }
 
+    fn op_summary_typed(name: &str, op_type: &str, conn: &CallConnection) -> OpSummary {
+        OpSummary {
+            name: name.to_string(),
+            schema: sample_schema_json(name, op_type),
+            connection: conn.clone(),
+            credentials_auth_token: None,
+        }
+    }
+
     #[test]
     fn build_bundles_same_peer_collision_returns_same_peer_collision_error() {
         let conn = CallConnection::new(stub_connection());
@@ -822,4 +872,234 @@ mod tests {
         assert_eq!(bundles.len(), 1);
         assert_eq!(bundles[0].spec.name, "worker/exec");
     }
+
+    // --- ADR-049 §8: streaming forwarding for Subscription ops -------------
+
+    #[test]
+    fn build_bundles_subscription_op_produces_stream_kind() {
+        let conn = CallConnection::new(stub_connection());
+        let discovered = vec![op_summary_typed("events/stream", "subscription", &conn)];
+        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
+        assert_eq!(bundles.len(), 1);
+        assert_eq!(bundles[0].spec.op_type, OperationType::Subscription);
+        assert!(
+            matches!(bundles[0].handler, HandlerKind::Stream(_)),
+            "Subscription op must register HandlerKind::Stream"
+        );
+        assert_eq!(bundles[0].provenance, OperationProvenance::FromCall);
+        assert!(bundles[0].composition_authority.is_none());
+        assert!(bundles[0].scoped_env.is_none());
+    }
+
+    #[test]
+    fn build_bundles_query_op_produces_once_kind() {
+        let conn = CallConnection::new(stub_connection());
+        let discovered = vec![op_summary_typed("fs/readFile", "query", &conn)];
+        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
+        assert_eq!(bundles.len(), 1);
+        assert_eq!(bundles[0].spec.op_type, OperationType::Query);
+        assert!(
+            matches!(bundles[0].handler, HandlerKind::Once(_)),
+            "Query op must register HandlerKind::Once"
+        );
+    }
+
+    #[test]
+    fn build_bundles_mutation_op_produces_once_kind() {
+        let conn = CallConnection::new(stub_connection());
+        let discovered = vec![op_summary_typed("fs/writeFile", "mutation", &conn)];
+        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
+        assert_eq!(bundles.len(), 1);
+        assert_eq!(bundles[0].spec.op_type, OperationType::Mutation);
+        assert!(
+            matches!(bundles[0].handler, HandlerKind::Once(_)),
+            "Mutation op must register HandlerKind::Once"
+        );
+    }
+
+    #[test]
+    fn build_bundles_mixed_op_types_route_to_correct_kind() {
+        let conn = CallConnection::new(stub_connection());
+        let discovered = vec![
+            op_summary_typed("fs/readFile", "query", &conn),
+            op_summary_typed("fs/writeFile", "mutation", &conn),
+            op_summary_typed("events/stream", "subscription", &conn),
+        ];
+        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
+        assert_eq!(bundles.len(), 3);
+        let by_name: std::collections::HashMap<&str, &HandlerKind> = bundles
+            .iter()
+            .map(|b| (b.spec.name.as_str(), &b.handler))
+            .collect();
+        assert!(matches!(by_name["fs/readFile"], HandlerKind::Once(_)));
+        assert!(matches!(by_name["fs/writeFile"], HandlerKind::Once(_)));
+        assert!(matches!(by_name["events/stream"], HandlerKind::Stream(_)));
+    }
+
+    /// Verify `make_streaming_forwarding_handler` produces a `StreamingHandler`
+    /// that builds the forwarded payload with `forwarded_for` populated from
+    /// `context.identity` (ADR-032) and calls `subscribe_with_payload`. Since
+    /// `subscribe_with_payload` on a mock connection returns a closed stream
+    /// (no transport), we capture the payload by intercepting the build step:
+    /// the handler's contract is "build payload via `build_forwarded_payload`,
+    /// then call `subscribe_with_payload(payload)`". We mirror the existing
+    /// `forwarding_handler_populates_forwarded_for` test by constructing the
+    /// handler and exercising the payload-construction path it relies on, plus
+    /// asserting the produced stream terminates (the mock-connection path
+    /// yields one error envelope then ends — no truncation, no hang).
+    #[tokio::test]
+    async fn streaming_forwarding_handler_populates_forwarded_for_and_streams() {
+        use futures::stream::StreamExt;
+
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
+        let captured = Arc::clone(&captured_payload);
+
+        let handler: StreamingHandler = {
+            let conn = Arc::clone(&conn);
+            make_streaming_handler(move |input, context| {
+                let conn = Arc::clone(&conn);
+                let captured = Arc::clone(&captured);
+                let remote_name = "events/stream".to_string();
+                use futures::stream::{once, StreamExt};
+                once(async move {
+                    let payload = build_forwarded_payload(&remote_name, input, &context, None);
+                    *captured.lock().unwrap() = Some(payload.clone());
+                    conn.subscribe_with_payload(payload).await
+                })
+                .flatten()
+            })
+        };
+
+        let ctx = test_context(Some(alice_identity()));
+        let mut stream = handler(json!({}), ctx);
+        let first = stream.next().await;
+        assert!(
+            first.is_some(),
+            "streaming forwarding handler must produce at least one envelope"
+        );
+        if let Some(env) = first {
+            assert!(
+                env.result.is_err(),
+                "mock connection has no transport, so the stream yields an error envelope"
+            );
+        }
+        let second = stream.next().await;
+        assert!(
+            second.is_none(),
+            "stream must terminate after the error (no truncation, no hang)"
+        );
+
+        let payload = captured_payload.lock().unwrap().clone().expect("captured");
+        assert_eq!(payload["operationId"], "events/stream");
+        assert_eq!(payload["forwarded_for"]["id"], "alice");
+    }
+
+    /// The streaming forwarding handler omits `forwarded_for` when
+    /// `context.identity` is `None`, mirroring the request/response handler.
+    #[tokio::test]
+    async fn streaming_forwarding_handler_omits_forwarded_for_when_identity_none() {
+        use futures::stream::StreamExt;
+
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
+        let captured = Arc::clone(&captured_payload);
+
+        let handler: StreamingHandler = {
+            let conn = Arc::clone(&conn);
+            make_streaming_handler(move |input, context| {
+                let conn = Arc::clone(&conn);
+                let captured = Arc::clone(&captured);
+                let remote_name = "events/stream".to_string();
+                use futures::stream::{once, StreamExt};
+                once(async move {
+                    let payload = build_forwarded_payload(&remote_name, input, &context, None);
+                    *captured.lock().unwrap() = Some(payload.clone());
+                    conn.subscribe_with_payload(payload).await
+                })
+                .flatten()
+            })
+        };
+
+        let ctx = test_context(None);
+        let mut stream = handler(json!({}), ctx);
+        let _ = stream.next().await;
+        let payload = captured_payload.lock().unwrap().clone().expect("captured");
+        assert!(
+            payload.get("forwarded_for").is_none(),
+            "forwarded_for must be omitted when context.identity is None"
+        );
+        assert_eq!(payload["operationId"], "events/stream");
+    }
+
+    /// The streaming forwarding handler populates `auth_token` when the hub's
+    /// own call-protocol token is provided.
+    #[tokio::test]
+    async fn streaming_forwarding_handler_sets_auth_token_when_provided() {
+        use futures::stream::StreamExt;
+
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
+        let captured = Arc::clone(&captured_payload);
+
+        let handler: StreamingHandler = {
+            let conn = Arc::clone(&conn);
+            make_streaming_handler(move |input, context| {
+                let conn = Arc::clone(&conn);
+                let captured = Arc::clone(&captured);
+                let remote_name = "events/stream".to_string();
+                use futures::stream::{once, StreamExt};
+                once(async move {
+                    let payload = build_forwarded_payload(
+                        &remote_name,
+                        input,
+                        &context,
+                        Some("alk_hub_token"),
+                    );
+                    *captured.lock().unwrap() = Some(payload.clone());
+                    conn.subscribe_with_payload(payload).await
+                })
+                .flatten()
+            })
+        };
+
+        let ctx = test_context(Some(alice_identity()));
+        let mut stream = handler(json!({}), ctx);
+        let _ = stream.next().await;
+        let payload = captured_payload.lock().unwrap().clone().expect("captured");
+        assert_eq!(payload["auth_token"], "alk_hub_token");
+        assert_eq!(payload["forwarded_for"]["id"], "alice");
+    }
+
+    /// `make_streaming_forwarding_handler` produces a `StreamingHandler` (not a
+    /// `Handler`) — verifies the helper returns the right type and that
+    /// `build_bundles` wires it into `HandlerKind::Stream`.
+    #[test]
+    fn make_streaming_forwarding_handler_returns_streaming_handler() {
+        let handler = make_streaming_forwarding_handler(
+            Arc::new(CallConnection::new(stub_connection())),
+            "events/stream".to_string(),
+            None,
+        );
+        let reg = HandlerRegistration::new(
+            OperationSpec::new(
+                "events/stream",
+                OperationType::Subscription,
+                Visibility::External,
+                json!({}),
+                json!({}),
+                vec![],
+                AccessControl::default(),
+            ),
+            HandlerKind::Stream(handler),
+            OperationProvenance::FromCall,
+            None,
+            None,
+            Capabilities::new(),
+        );
+        assert!(matches!(reg.handler, HandlerKind::Stream(_)));
+        assert_eq!(reg.provenance, OperationProvenance::FromCall);
+        assert!(reg.composition_authority.is_none());
+        assert!(reg.scoped_env.is_none());
+    }
 }

crates/alknet-call/src/client/from_jsonschema.rs

@@ -11,7 +11,9 @@ use serde_json::Value;
 use crate::client::{AdapterError, OperationAdapter};
 use crate::protocol::wire::{CallError, ResponseEnvelope};
 use crate::registry::context::OperationContext;
-use crate::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+use crate::registry::registration::{
+    make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+};
 use crate::registry::spec::OperationSpec;
 
 /// Build a [`HandlerRegistration`] from a JSON Schema-described operation.
@@ -30,7 +32,7 @@ pub fn from_jsonschema(spec: OperationSpec, _schema: Value) -> HandlerRegistrati
     });
     HandlerRegistration::new(
         spec,
-        handler,
+        HandlerKind::Once(handler),
         OperationProvenance::FromJsonSchema,
         None,
         None,
@@ -138,7 +140,10 @@ mod tests {
     async fn placeholder_handler_returns_error_when_invoked() {
         let bundle = from_jsonschema_fn::from_jsonschema(test_spec("ns/op"), serde_json::json!({}));
         let ctx = test_context("req-1");
-        let response = (bundle.handler)(serde_json::json!({}), ctx).await;
+        let response = match &bundle.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Err(e) => {
                 assert_eq!(e.code, "NOT_FOUND");

crates/alknet-call/src/protocol/adapter.rs

@@ -166,7 +166,9 @@ mod tests {
     };
     use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
     use crate::registry::env::OperationEnv;
-    use crate::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+    use crate::registry::registration::{
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    };
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::AuthToken;
     use alknet_core::types::Capabilities;
@@ -245,7 +247,8 @@ mod tests {
         handler: crate::registry::registration::Handler,
     ) -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     name,
                     OperationType::Query,
@@ -255,12 +258,13 @@ mod tests {
                     vec![],
                     acl,
                 ),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -421,14 +425,16 @@ mod tests {
         let mut registry = OperationRegistry::new();
         let scoped = ScopedPeerEnv::new(["fs/readFile"]);
         let caps = Capabilities::new().with_api_key("google", "k".to_string());
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("agent/run", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 Some(scoped.clone()),
                 caps.clone(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let adapter = CallAdapter::new(registry, provider);
@@ -543,7 +549,7 @@ mod tests {
                 vec![],
                 AccessControl::default(),
             ),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::FromCall,
             None,
             None,
@@ -610,7 +616,7 @@ mod tests {
                 vec![],
                 AccessControl::default(),
             ),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::FromCall,
             None,
             None,

crates/alknet-call/src/protocol/connection.rs

@@ -26,7 +26,7 @@ use super::wire::{
 use crate::protocol::wire::ResponseEnvelope;
 use crate::registry::context::{generate_request_id, AbortPolicy, OperationContext, ScopedPeerEnv};
 use crate::registry::env::OperationEnv;
-use crate::registry::registration::{Handler, HandlerRegistration};
+use crate::registry::registration::{HandlerKind, HandlerRegistration};
 use crate::registry::spec::AccessResult;
 
 const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(30);
@@ -168,11 +168,26 @@ impl CallConnection {
         operation_id: &str,
         input: Value,
     ) -> impl Stream<Item = ResponseEnvelope> {
-        let request_id = generate_request_id();
         let payload = serde_json::json!({
             "operationId": operation_id,
             "input": input,
         });
+        self.subscribe_with_payload(payload).await
+    }
+
+    /// Subscribe to a remote op with a caller-constructed `call.requested`
+    /// payload. The payload MUST include `operationId` and `input`; the
+    /// caller may add `forwarded_for` (ADR-032) and `auth_token` (ADR-017 §7)
+    /// for the hub forwarding path used by `from_call`'s streaming forwarding
+    /// handler. Mirrors [`call_with_payload`](Self::call_with_payload) so the
+    /// forwarding handler can populate `forwarded_for` + `auth_token` on the
+    /// subscription payload (the plain [`subscribe`](Self::subscribe) builds
+    /// the payload internally and omits those fields).
+    pub async fn subscribe_with_payload(
+        &self,
+        payload: Value,
+    ) -> impl Stream<Item = ResponseEnvelope> {
+        let request_id = generate_request_id();
 
         let connection = match &self.connection {
             Some(c) => c,
@@ -307,7 +322,7 @@ impl OperationEnv for OverlayOperationEnv {
             return ResponseEnvelope::not_found(parent.request_id.clone(), &name);
         }
 
-        let handler: Handler;
+        let handler: HandlerKind;
         let composition_authority;
         let scoped_env;
         let access_control;
@@ -316,7 +331,7 @@ impl OperationEnv for OverlayOperationEnv {
             let Some(registration) = overlay.get(&name) else {
                 return ResponseEnvelope::not_found(parent.request_id.clone(), &name);
             };
-            handler = Arc::clone(&registration.handler);
+            handler = registration.handler.clone();
             composition_authority = registration.composition_authority.clone();
             scoped_env = registration
                 .scoped_env
@@ -355,7 +370,15 @@ impl OperationEnv for OverlayOperationEnv {
             internal: true,
         };
 
-        handler(input, context).await
+        match handler {
+            HandlerKind::Once(h) => h(input, context).await,
+            HandlerKind::Stream(_) => ResponseEnvelope::error(
+                parent.request_id.clone(),
+                CallError::invalid_operation_type(
+                    "OperationEnv::invoke() called on a Subscription op; composition is request/response-only",
+                ),
+            ),
+        }
     }
 
     fn contains(&self, name: &str) -> bool {
@@ -421,7 +444,7 @@ impl Stream for SubscriptionStream {
 mod tests {
     use super::*;
     use crate::registry::context::CompositionAuthority;
-    use crate::registry::registration::{make_handler, OperationProvenance};
+    use crate::registry::registration::{make_handler, Handler, HandlerKind, OperationProvenance};
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::types::{Capabilities, MockConnection};
     use std::collections::HashMap;
@@ -476,7 +499,7 @@ mod tests {
     fn imported_registration(name: &str) -> HandlerRegistration {
         HandlerRegistration::new(
             external_spec(name),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::FromCall,
             None,
             None,
@@ -608,7 +631,7 @@ mod tests {
         });
         conn.register_imported(HandlerRegistration::new(
             external_spec("worker/exec"),
-            inspect_handler,
+            HandlerKind::Once(inspect_handler),
             OperationProvenance::FromCall,
             None,
             None,
@@ -631,7 +654,9 @@ mod tests {
     fn connection_accessor_returns_underlying_connection() {
         let conn = CallConnection::new(stub_connection());
         assert_eq!(
-            conn.connection().expect("quic connection present").remote_alpn(),
+            conn.connection()
+                .expect("quic connection present")
+                .remote_alpn(),
             b"alknet/call"
         );
     }
@@ -960,4 +985,39 @@ mod tests {
         assert!(conn.connection().is_some(), "QUIC connection present");
         assert!(conn.identity().is_none(), "no identity set yet");
     }
+
+    #[tokio::test]
+    async fn overlay_env_invoke_on_stream_kind_returns_invalid_operation_type() {
+        use crate::registry::registration::make_streaming_handler;
+        let conn = CallConnection::new(stub_connection());
+        let streaming_handler = make_streaming_handler(|input, ctx| {
+            futures::stream::iter(vec![ResponseEnvelope::ok(ctx.request_id, input)])
+        });
+        conn.register_imported(HandlerRegistration::new(
+            OperationSpec::new(
+                "events/stream",
+                OperationType::Subscription,
+                Visibility::External,
+                serde_json::json!({}),
+                serde_json::json!({}),
+                vec![],
+                AccessControl::default(),
+            ),
+            HandlerKind::Stream(streaming_handler),
+            OperationProvenance::FromCall,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        let env = conn.overlay_env();
+        let scoped = ScopedPeerEnv::new(["events/stream"]);
+        let ctx = root_context("root-stream", scoped, env.clone());
+        let response = env
+            .invoke("events", "stream", serde_json::json!({}), &ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
 }

crates/alknet-call/src/protocol/dispatch.rs

@@ -326,7 +326,9 @@ impl Clone for Dispatcher {
 mod tests {
     use super::*;
     use crate::protocol::wire::EVENT_RESPONDED;
-    use crate::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+    use crate::registry::registration::{
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    };
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
     use alknet_core::types::{Capabilities, MockConnection};
@@ -412,7 +414,8 @@ mod tests {
 
     fn registry_with(name: &str, visibility: Visibility, acl: AccessControl) -> OperationRegistry {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     name,
                     OperationType::Query,
@@ -422,14 +425,15 @@ mod tests {
                     vec![],
                     acl,
                 ),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         registry
     }
 
@@ -451,14 +455,16 @@ mod tests {
                 serde_json::json!({ "has_google": has_google }),
             )
         });
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("admin/run", AccessControl::default()),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 caps,
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = Dispatcher::new(registry, provider);
@@ -486,7 +492,8 @@ mod tests {
                 serde_json::json!({ "has_google": has_google }),
             )
         });
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin/run",
                     AccessControl {
@@ -494,12 +501,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 caps,
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(
             StaticIdentityProvider::new()
@@ -609,14 +617,16 @@ mod tests {
                 serde_json::json!({ "forwarded_for_id": forwarded_id }),
             )
         });
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("fs/readFile", AccessControl::default()),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = Dispatcher::new(registry, provider);
@@ -648,14 +658,16 @@ mod tests {
                 serde_json::json!({ "present": present }),
             )
         });
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("fs/readFile", AccessControl::default()),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = Dispatcher::new(registry, provider);
@@ -736,14 +748,16 @@ mod tests {
                 serde_json::json!({ "peer_ids": peer_ids }),
             )
         });
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("fs/readFile", AccessControl::default()),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = Dispatcher::new(registry, provider);
@@ -795,7 +809,11 @@ mod tests {
         let child_id = "ws-abort-child".to_string();
         {
             let mut pending = conn.pending().lock();
-            pending.register_call(parent_id.clone(), Instant::now() + Duration::from_secs(30), None);
+            pending.register_call(
+                parent_id.clone(),
+                Instant::now() + Duration::from_secs(30),
+                None,
+            );
             pending.register_call(
                 child_id.clone(),
                 Instant::now() + Duration::from_secs(30),
@@ -844,11 +862,16 @@ mod tests {
             "input": { "v": 42 },
         });
         let request_id = "ws-roundtrip-1".to_string();
-        let response = dp.dispatch_requested(&conn, request_id.clone(), payload).await;
+        let response = dp
+            .dispatch_requested(&conn, request_id.clone(), payload)
+            .await;
         assert!(response.result.is_ok());
         let envelope: EventEnvelope = response.into();
         assert_eq!(envelope.r#type, EVENT_RESPONDED);
         assert_eq!(envelope.id, "ws-roundtrip-1");
-        assert_eq!(envelope.payload.get("output"), Some(&serde_json::json!({ "v": 42 })));
+        assert_eq!(
+            envelope.payload.get("output"),
+            Some(&serde_json::json!({ "v": 42 }))
+        );
     }
 }

crates/alknet-call/src/protocol/wire.rs

@@ -105,6 +105,10 @@ impl CallError {
     pub fn timeout(message: impl Into<String>) -> Self {
         Self::new("TIMEOUT", message, true)
     }
+
+    pub fn invalid_operation_type(message: impl Into<String>) -> Self {
+        Self::new("INVALID_OPERATION_TYPE", message, false)
+    }
 }
 
 impl Eq for CallError {}

crates/alknet-call/src/registry/discovery.rs

@@ -324,7 +324,10 @@ pub fn services_schema_handler(registry: Arc<OperationRegistry>) -> Handler {
 mod tests {
     use super::*;
     use crate::registry::context::{CompositionAuthority, ScopedPeerEnv};
-    use crate::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+    use crate::registry::registration::{
+        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
+        OperationProvenance, StreamingHandler,
+    };
     use alknet_core::types::Capabilities;
     use std::collections::HashMap;
     use std::time::Duration;
@@ -359,6 +362,12 @@ mod tests {
         )
     }
 
+    fn echo_streaming_handler() -> StreamingHandler {
+        make_streaming_handler(|input, context| {
+            futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
+        })
+    }
+
     fn noop_env() -> Arc<dyn crate::registry::env::OperationEnv + Send + Sync> {
         struct NoopEnv;
         #[async_trait::async_trait]
@@ -439,15 +448,18 @@ mod tests {
 
     fn registry_with_access_controlled_ops() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec_with_acl("public/echo", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 external_spec_with_acl(
                     "admin/secret",
                     AccessControl {
@@ -455,20 +467,23 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("internal/hidden"),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -485,23 +500,28 @@ mod tests {
 
     fn registry_with_ops() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("fs/readFile"),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret/internal"),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     "events/subscribe",
                     OperationType::Subscription,
@@ -511,13 +531,15 @@ mod tests {
                     vec![],
                     AccessControl::default(),
                 ),
-            echo_handler(),
+                HandlerKind::Stream(echo_streaming_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     "fs/readFileErr",
                     OperationType::Query,
@@ -532,12 +554,13 @@ mod tests {
                     }],
                     AccessControl::default(),
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -669,22 +692,26 @@ mod tests {
         let schema_handler = services_schema_handler(Arc::clone(&registry));
 
         let mut discovery_registry = OperationRegistry::new();
-        discovery_registry.register(HandlerRegistration::new(
+        discovery_registry
+            .register(HandlerRegistration::new(
                 services_list_spec(),
-            list_handler,
+                HandlerKind::Once(list_handler),
                 OperationProvenance::Local,
                 CompositionAuthority::none(),
                 ScopedPeerEnv::empty().into(),
                 Capabilities::new(),
-        ));
+            ))
-        discovery_registry.register(HandlerRegistration::new(
+            .unwrap();
+        discovery_registry
+            .register(HandlerRegistration::new(
                 services_schema_spec(),
-            schema_handler,
+                HandlerKind::Once(schema_handler),
                 OperationProvenance::Local,
                 CompositionAuthority::none(),
                 ScopedPeerEnv::empty().into(),
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let discovery = Arc::new(discovery_registry);
 
         let ctx = root_context("req-6");

crates/alknet-call/src/registry/env.rs

@@ -303,7 +303,9 @@ impl OperationEnv for PeerCompositeEnv {
 mod tests {
     use super::*;
     use crate::registry::context::CompositionAuthority;
-    use crate::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+    use crate::registry::registration::{
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    };
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::Identity;
     use alknet_core::types::Capabilities;
@@ -406,7 +408,8 @@ mod tests {
         scoped_env: Option<ScopedPeerEnv>,
     ) -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     name,
                     OperationType::Query,
@@ -416,12 +419,13 @@ mod tests {
                     vec![],
                     AccessControl::default(),
                 ),
-            handler,
+                HandlerKind::Once(handler),
                 OperationProvenance::Local,
                 composition_authority,
                 scoped_env,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 

crates/alknet-call/src/registry/registration.rs

@@ -4,11 +4,12 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use alknet_core::types::Capabilities;
+use futures::stream::{self, Stream};
 use serde_json::Value;
 
 use super::context::{CompositionAuthority, OperationContext, ScopedPeerEnv};
-use super::spec::{AccessResult, OperationSpec, Visibility};
+use super::spec::{AccessResult, OperationSpec, OperationType, Visibility};
-use crate::protocol::wire::ResponseEnvelope;
+use crate::protocol::wire::{CallError, ResponseEnvelope};
 
 pub type Handler = Arc<
     dyn Fn(Value, OperationContext) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
@@ -16,6 +17,20 @@ pub type Handler = Arc<
         + Sync,
 >;
 
+pub type StreamingHandler = Arc<
+    dyn Fn(Value, OperationContext) -> Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>
+        + Send
+        + Sync,
+>;
+
+pub type ResponseStream = Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>;
+
+#[derive(Clone)]
+pub enum HandlerKind {
+    Once(Handler),
+    Stream(StreamingHandler),
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum OperationProvenance {
     Local,
@@ -28,7 +43,7 @@ pub enum OperationProvenance {
 
 pub struct HandlerRegistration {
     pub spec: OperationSpec,
-    pub handler: Handler,
+    pub handler: HandlerKind,
     pub provenance: OperationProvenance,
     pub composition_authority: Option<CompositionAuthority>,
     pub scoped_env: Option<ScopedPeerEnv>,
@@ -38,7 +53,7 @@ pub struct HandlerRegistration {
 impl HandlerRegistration {
     pub fn new(
         spec: OperationSpec,
-        handler: Handler,
+        handler: HandlerKind,
         provenance: OperationProvenance,
         composition_authority: Option<CompositionAuthority>,
         scoped_env: Option<ScopedPeerEnv>,
@@ -66,9 +81,24 @@ impl OperationRegistry {
         }
     }
 
-    pub fn register(&mut self, registration: HandlerRegistration) {
+    pub fn register(&mut self, registration: HandlerRegistration) -> Result<(), String> {
+        let expected = match registration.spec.op_type {
+            OperationType::Query | OperationType::Mutation => "Once",
+            OperationType::Subscription => "Stream",
+        };
+        let actual = match registration.handler {
+            HandlerKind::Once(_) => "Once",
+            HandlerKind::Stream(_) => "Stream",
+        };
+        if expected != actual {
+            return Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::{} (got HandlerKind::{})",
+                registration.spec.op_type, expected, actual
+            ));
+        }
         self.operations
             .insert(registration.spec.name.clone(), registration);
+        Ok(())
     }
 
     pub fn registration(&self, name: &str) -> Option<&HandlerRegistration> {
@@ -113,9 +143,76 @@ impl OperationRegistry {
             return ResponseEnvelope::forbidden(request_id, message);
         }
 
-        let handler = Arc::clone(&registration.handler);
+        match &registration.handler {
+            HandlerKind::Once(handler) => {
+                let handler = Arc::clone(handler);
                 (handler)(input, context).await
             }
+            HandlerKind::Stream(_) => ResponseEnvelope::error(
+                request_id,
+                CallError::invalid_operation_type(
+                    "invoke() called on a Subscription op; use invoke_streaming()",
+                ),
+            ),
+        }
+    }
+
+    pub fn invoke_streaming(
+        &self,
+        name: &str,
+        input: Value,
+        context: OperationContext,
+    ) -> ResponseStream {
+        let request_id = context.request_id.clone();
+        let name_owned = name.to_string();
+
+        let registration = match self.operations.get(name) {
+            Some(r) => r,
+            None => {
+                return Box::pin(stream::once(async move {
+                    ResponseEnvelope::not_found(request_id, &name_owned)
+                }));
+            }
+        };
+
+        if registration.spec.visibility == Visibility::Internal && !context.internal {
+            return Box::pin(stream::once(async move {
+                ResponseEnvelope::not_found(request_id, &name_owned)
+            }));
+        }
+
+        let acl = &registration.spec.access_control;
+        let identity = if context.internal {
+            context
+                .handler_identity
+                .as_ref()
+                .and_then(|ca| ca.as_identity())
+        } else {
+            context.identity.clone()
+        };
+
+        if let AccessResult::Forbidden(message) = acl.check(identity.as_ref()) {
+            return Box::pin(stream::once(async move {
+                ResponseEnvelope::forbidden(request_id, message)
+            }));
+        }
+
+        let streaming_handler = match &registration.handler {
+            HandlerKind::Stream(h) => Arc::clone(h),
+            HandlerKind::Once(_) => {
+                return Box::pin(stream::once(async move {
+                    ResponseEnvelope::error(
+                        request_id,
+                        CallError::invalid_operation_type(
+                            "invoke_streaming() called on a Query/Mutation op; use invoke()",
+                        ),
+                    )
+                }));
+            }
+        };
+
+        streaming_handler(input, context)
+    }
 }
 
 impl Default for OperationRegistry {
@@ -135,10 +232,30 @@ impl OperationRegistryBuilder {
         }
     }
 
-    fn store(mut self, registration: HandlerRegistration) -> Self {
+    fn store(mut self, registration: HandlerRegistration) -> Result<Self, String> {
-        self.operations
+        let name = registration.spec.name.clone();
-            .insert(registration.spec.name.clone(), registration);
+        self.operations.insert(name, registration);
-        self
+        Ok(self)
+    }
+
+    fn wrap_once(spec: &OperationSpec, handler: Handler) -> Result<HandlerKind, String> {
+        match spec.op_type {
+            OperationType::Query | OperationType::Mutation => Ok(HandlerKind::Once(handler)),
+            OperationType::Subscription => Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::Stream (got Handler)",
+                spec.op_type
+            )),
+        }
+    }
+
+    fn wrap_stream(spec: &OperationSpec, handler: StreamingHandler) -> Result<HandlerKind, String> {
+        match spec.op_type {
+            OperationType::Subscription => Ok(HandlerKind::Stream(handler)),
+            OperationType::Query | OperationType::Mutation => Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::Once (got StreamingHandler)",
+                spec.op_type
+            )),
+        }
     }
 
     pub fn with_local(
@@ -148,10 +265,31 @@ impl OperationRegistryBuilder {
         composition_authority: Option<CompositionAuthority>,
         scoped_env: Option<ScopedPeerEnv>,
         capabilities: Capabilities,
-    ) -> Self {
+    ) -> Result<Self, String> {
+        let kind = Self::wrap_once(&spec, handler)?;
         let registration = HandlerRegistration::new(
             spec,
-            handler,
+            kind,
+            OperationProvenance::Local,
+            composition_authority,
+            scoped_env,
+            capabilities,
+        );
+        self.store(registration)
+    }
+
+    pub fn with_local_streaming(
+        self,
+        spec: OperationSpec,
+        handler: StreamingHandler,
+        composition_authority: Option<CompositionAuthority>,
+        scoped_env: Option<ScopedPeerEnv>,
+        capabilities: Capabilities,
+    ) -> Result<Self, String> {
+        let kind = Self::wrap_stream(&spec, handler)?;
+        let registration = HandlerRegistration::new(
+            spec,
+            kind,
             OperationProvenance::Local,
             composition_authority,
             scoped_env,
@@ -165,7 +303,7 @@ impl OperationRegistryBuilder {
         spec: OperationSpec,
         handler: Handler,
         capabilities: Capabilities,
-    ) -> Self {
+    ) -> Result<Self, String> {
         self.with_leaf_provenance(
             spec,
             handler,
@@ -180,13 +318,41 @@ impl OperationRegistryBuilder {
         handler: Handler,
         provenance: OperationProvenance,
         capabilities: Capabilities,
-    ) -> Self {
+    ) -> Result<Self, String> {
+        let kind = Self::wrap_once(&spec, handler)?;
         let registration =
-            HandlerRegistration::new(spec, handler, provenance, None, None, capabilities);
+            HandlerRegistration::new(spec, kind, provenance, None, None, capabilities);
         self.store(registration)
     }
 
-    pub fn with(self, registration: HandlerRegistration) -> Self {
+    pub fn with_leaf_streaming(
+        self,
+        spec: OperationSpec,
+        handler: StreamingHandler,
+        capabilities: Capabilities,
+    ) -> Result<Self, String> {
+        self.with_leaf_streaming_provenance(
+            spec,
+            handler,
+            OperationProvenance::FromOpenAPI,
+            capabilities,
+        )
+    }
+
+    pub fn with_leaf_streaming_provenance(
+        self,
+        spec: OperationSpec,
+        handler: StreamingHandler,
+        provenance: OperationProvenance,
+        capabilities: Capabilities,
+    ) -> Result<Self, String> {
+        let kind = Self::wrap_stream(&spec, handler)?;
+        let registration =
+            HandlerRegistration::new(spec, kind, provenance, None, None, capabilities);
+        self.store(registration)
+    }
+
+    pub fn with(self, registration: HandlerRegistration) -> Result<Self, String> {
         self.store(registration)
     }
 
@@ -211,6 +377,14 @@ where
     Arc::new(move |input, context| Box::pin(f(input, context)))
 }
 
+pub fn make_streaming_handler<S, St>(f: S) -> StreamingHandler
+where
+    S: Fn(Value, OperationContext) -> St + Send + Sync + 'static,
+    St: Stream<Item = ResponseEnvelope> + Send + 'static,
+{
+    Arc::new(move |input, context| Box::pin(f(input, context)))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -312,14 +486,16 @@ mod tests {
     #[tokio::test]
     async fn register_and_invoke_simple_operation() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context("req-1", None, None, false, ScopedPeerEnv::empty());
         let response = registry
             .invoke("echo", serde_json::json!({"hi": 1}), ctx)
@@ -331,14 +507,16 @@ mod tests {
     #[tokio::test]
     async fn internal_op_from_external_call_returns_not_found() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context("req-2", None, None, false, ScopedPeerEnv::empty());
         let response = registry.invoke("secret", serde_json::json!({}), ctx).await;
         match response.result {
@@ -353,14 +531,16 @@ mod tests {
     #[tokio::test]
     async fn internal_op_from_internal_call_invokes_handler() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context("req-3", None, None, true, ScopedPeerEnv::empty());
         let response = registry
             .invoke("secret", serde_json::json!({"x": 2}), ctx)
@@ -383,7 +563,8 @@ mod tests {
     #[tokio::test]
     async fn acl_sufficient_scopes_allowed() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin",
                     AccessControl {
@@ -391,12 +572,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context(
             "req-5",
             Some(identity_with_scopes("caller", &["admin"])),
@@ -411,7 +593,8 @@ mod tests {
     #[tokio::test]
     async fn acl_insufficient_scopes_forbidden() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin",
                     AccessControl {
@@ -419,12 +602,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context(
             "req-6",
             Some(identity_with_scopes("caller", &["user"])),
@@ -445,7 +629,8 @@ mod tests {
     #[tokio::test]
     async fn acl_restricted_op_no_identity_forbidden() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin",
                     AccessControl {
@@ -453,12 +638,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context("req-7", None, None, false, ScopedPeerEnv::empty());
         let response = registry.invoke("admin", serde_json::json!({}), ctx).await;
         match response.result {
@@ -474,7 +660,8 @@ mod tests {
     async fn internal_call_acl_uses_handler_identity() {
         let mut registry = OperationRegistry::new();
         let composing_authority = CompositionAuthority::new("agent-chat", ["admin".to_string()]);
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec(
                     "secret",
                     AccessControl {
@@ -482,12 +669,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context(
             "req-8",
             Some(identity_with_scopes("user", &["user"])),
@@ -506,7 +694,8 @@ mod tests {
     async fn internal_call_acl_insufficient_handler_identity_forbidden() {
         let mut registry = OperationRegistry::new();
         let weak_authority = CompositionAuthority::new("weak", ["user".to_string()]);
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec(
                     "secret",
                     AccessControl {
@@ -514,12 +703,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context(
             "req-9",
             Some(identity_with_scopes("user", &["admin"])),
@@ -541,7 +731,8 @@ mod tests {
     async fn external_call_acl_uses_caller_identity_not_handler_identity() {
         let mut registry = OperationRegistry::new();
         let handler_authority = CompositionAuthority::new("agent", ["admin".to_string()]);
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "gate",
                     AccessControl {
@@ -549,12 +740,13 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 Some(handler_authority),
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context(
             "req-10",
             Some(identity_with_scopes("user", &["user"])),
@@ -572,22 +764,26 @@ mod tests {
     #[tokio::test]
     async fn list_operations_returns_external_only() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ops = registry.list_operations();
         assert_eq!(ops.len(), 1);
         assert_eq!(ops[0].name, "echo");
@@ -596,14 +792,16 @@ mod tests {
     #[tokio::test]
     async fn handler_returned_error_passes_through() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("boom", AccessControl::default()),
-            error_handler(),
+                HandlerKind::Once(error_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let ctx = root_context("req-11", None, None, false, ScopedPeerEnv::empty());
         let response = registry.invoke("boom", serde_json::json!({}), ctx).await;
         match response.result {
@@ -622,6 +820,7 @@ mod tests {
                 ScopedPeerEnv::empty().into(),
                 Capabilities::new(),
             )
+            .unwrap()
             .build();
         let reg = registry.registration("echo").expect("registered");
         assert_eq!(reg.provenance, OperationProvenance::Local);
@@ -639,6 +838,7 @@ mod tests {
                 Some(ScopedPeerEnv::new(["fs/readFile"])),
                 Capabilities::new(),
             )
+            .unwrap()
             .build();
         let reg = registry.registration("agent").expect("registered");
         assert_eq!(reg.provenance, OperationProvenance::Local);
@@ -657,6 +857,7 @@ mod tests {
                 echo_handler(),
                 Capabilities::new(),
             )
+            .unwrap()
             .build();
         let reg = registry.registration("vastai").expect("registered");
         assert_eq!(reg.provenance, OperationProvenance::FromOpenAPI);
@@ -673,6 +874,7 @@ mod tests {
                 OperationProvenance::FromCall,
                 Capabilities::new(),
             )
+            .unwrap()
             .build();
         let reg = registry.registration("remote").expect("registered");
         assert_eq!(reg.provenance, OperationProvenance::FromCall);
@@ -684,13 +886,16 @@ mod tests {
     fn builder_with_takes_full_bundle() {
         let registration = HandlerRegistration::new(
             external_spec("agent", AccessControl::default()),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Session,
             Some(CompositionAuthority::new("sandbox", [])),
             Some(ScopedPeerEnv::new(["fs/readFile"])),
             Capabilities::new(),
         );
-        let registry = OperationRegistryBuilder::new().with(registration).build();
+        let registry = OperationRegistryBuilder::new()
+            .with(registration)
+            .unwrap()
+            .build();
         let reg = registry.registration("agent").expect("registered");
         assert_eq!(reg.provenance, OperationProvenance::Session);
         assert!(reg.composition_authority.is_some());
@@ -717,18 +922,330 @@ mod tests {
         let authority = CompositionAuthority::new("agent", ["fs:read".to_string()]);
         let scoped = ScopedPeerEnv::new(["fs/readFile"]);
         let caps = Capabilities::new().with_api_key("google", "k".to_string());
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("agent", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 Some(authority.clone()),
                 Some(scoped.clone()),
                 caps.clone(),
-        ));
+            ))
+            .unwrap();
         let reg = registry.registration("agent").expect("found");
         assert_eq!(reg.spec.name, "agent");
         assert_eq!(reg.provenance, OperationProvenance::Local);
         assert_eq!(reg.composition_authority.as_ref().unwrap().label, "agent");
         assert!(reg.scoped_env.as_ref().unwrap().allows("fs/readFile"));
     }
+
+    fn subscription_spec(name: &str) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Subscription,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            AccessControl::default(),
+        )
+    }
+
+    fn echo_streaming_handler() -> StreamingHandler {
+        make_streaming_handler(|input, context| {
+            futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
+        })
+    }
+
+    #[tokio::test]
+    async fn invoke_on_stream_kind_returns_invalid_operation_type() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec("events/stream"),
+                HandlerKind::Stream(echo_streaming_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-iot", None, None, false, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke("events/stream", serde_json::json!({}), ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_on_once_kind_dispatches_normally() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                external_spec("echo", AccessControl::default()),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-once", None, None, false, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke("echo", serde_json::json!({"hi": 1}), ctx)
+            .await;
+        assert_eq!(response.result, Ok(serde_json::json!({"hi": 1})));
+    }
+
+    #[test]
+    fn register_rejects_once_for_subscription_spec() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            subscription_spec("events/stream"),
+            HandlerKind::Once(echo_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        match result {
+            Err(msg) => assert!(
+                msg.contains("Subscription")
+                    && msg.contains("HandlerKind::Stream")
+                    && msg.contains("HandlerKind::Once"),
+                "unexpected message: {msg}"
+            ),
+            other => panic!("expected Err, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn register_rejects_stream_for_query_spec() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            external_spec("echo", AccessControl::default()),
+            HandlerKind::Stream(echo_streaming_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        match result {
+            Err(msg) => assert!(
+                (msg.contains("Query") || msg.contains("Mutation"))
+                    && msg.contains("HandlerKind::Once")
+                    && msg.contains("HandlerKind::Stream"),
+                "unexpected message: {msg}"
+            ),
+            other => panic!("expected Err, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn make_streaming_handler_produces_working_stream() {
+        use futures::stream::StreamExt;
+        let handler = echo_streaming_handler();
+        let ctx = root_context("req-st", None, None, false, ScopedPeerEnv::empty());
+        let mut stream = handler(serde_json::json!({"v": 1}), ctx);
+        let first = stream.next().await.expect("one envelope");
+        assert_eq!(first.result, Ok(serde_json::json!({"v": 1})));
+        let second = stream.next().await;
+        assert!(second.is_none(), "stream ends after one value");
+    }
+
+    #[test]
+    fn call_error_invalid_operation_type_is_not_retryable() {
+        let err = CallError::invalid_operation_type("bad path");
+        assert_eq!(err.code, "INVALID_OPERATION_TYPE");
+        assert!(!err.retryable);
+        assert!(err.details.is_none());
+    }
+
+    async fn collect_stream(mut s: ResponseStream) -> Vec<ResponseEnvelope> {
+        use futures::stream::StreamExt;
+        let mut out = Vec::new();
+        while let Some(env) = s.next().await {
+            out.push(env);
+        }
+        out
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_subscription_dispatches_handler_stream() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec("events/stream"),
+                HandlerKind::Stream(echo_streaming_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-is-1", None, None, false, ScopedPeerEnv::empty());
+        let stream = registry.invoke_streaming("events/stream", serde_json::json!({"v": 7}), ctx);
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        assert_eq!(items[0].request_id, "req-is-1");
+        assert_eq!(items[0].result, Ok(serde_json::json!({"v": 7})));
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_unknown_op_yields_single_not_found() {
+        let registry = OperationRegistry::new();
+        let ctx = root_context("req-is-2", None, None, false, ScopedPeerEnv::empty());
+        let stream = registry.invoke_streaming("missing", serde_json::json!({}), ctx);
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => {
+                assert_eq!(e.code, "NOT_FOUND");
+                assert!(e.message.contains("missing"));
+            }
+            other => panic!("expected NOT_FOUND, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_query_op_yields_invalid_operation_type() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                external_spec("echo", AccessControl::default()),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-is-3", None, None, false, ScopedPeerEnv::empty());
+        let stream = registry.invoke_streaming("echo", serde_json::json!({}), ctx);
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_internal_op_from_external_yields_not_found() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                internal_subscription_spec(AccessControl::default()),
+                HandlerKind::Stream(echo_streaming_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-is-4", None, None, false, ScopedPeerEnv::empty());
+        let stream = registry.invoke_streaming("events/stream", serde_json::json!({}), ctx);
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => {
+                assert_eq!(e.code, "NOT_FOUND");
+                assert!(e.message.contains("events/stream"));
+            }
+            other => panic!("expected NOT_FOUND, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_acl_denied_yields_forbidden() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec_with_acl(AccessControl {
+                    required_scopes: vec!["admin".to_string()],
+                    ..Default::default()
+                }),
+                HandlerKind::Stream(echo_streaming_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context(
+            "req-is-5",
+            Some(identity_with_scopes("caller", &["user"])),
+            None,
+            false,
+            ScopedPeerEnv::empty(),
+        );
+        let stream = registry.invoke_streaming("events/stream", serde_json::json!({}), ctx);
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => {
+                assert_eq!(e.code, "FORBIDDEN");
+                assert!(e.message.contains("admin"));
+            }
+            other => panic!("expected FORBIDDEN, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_internal_call_uses_handler_identity_for_acl() {
+        let mut registry = OperationRegistry::new();
+        let composing_authority = CompositionAuthority::new("agent-chat", ["admin".to_string()]);
+        registry
+            .register(HandlerRegistration::new(
+                internal_subscription_spec(AccessControl {
+                    required_scopes: vec!["admin".to_string()],
+                    ..Default::default()
+                }),
+                HandlerKind::Stream(echo_streaming_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context(
+            "req-is-6",
+            Some(identity_with_scopes("user", &["user"])),
+            Some(composing_authority),
+            true,
+            ScopedPeerEnv::empty(),
+        );
+        let stream = registry.invoke_streaming("events/stream", serde_json::json!({"ok": 1}), ctx);
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        assert_eq!(items[0].request_id, "req-is-6");
+        assert_eq!(items[0].result, Ok(serde_json::json!({"ok": 1})));
+    }
+
+    fn subscription_spec_with_acl(acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            "events/stream",
+            OperationType::Subscription,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+        )
+    }
+
+    fn internal_subscription_spec(acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            "events/stream",
+            OperationType::Subscription,
+            Visibility::Internal,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+        )
+    }
 }

crates/alknet-call/tests/two_node_call.rs

@@ -15,7 +15,7 @@ use alknet_call::registry::discovery::{
     services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
 };
 use alknet_call::registry::registration::{
-    make_handler, Handler, HandlerRegistration, OperationProvenance, OperationRegistry,
+    make_handler, Handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
 };
 use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
 use alknet_core::auth::{Identity, IdentityProvider};
@@ -124,58 +124,66 @@ async fn build_raw_quinn_server(
 /// services/list + services/schema discovery handlers.
 fn build_server_registry() -> Arc<OperationRegistry> {
     let mut registry = OperationRegistry::new();
-    registry.register(HandlerRegistration::new(
+    registry
+        .register(HandlerRegistration::new(
             external_spec("server/echo"),
-        echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Local,
             None,
             None,
             Capabilities::new(),
-    ));
+        ))
-    registry.register(HandlerRegistration::new(
+        .unwrap();
+    registry
+        .register(HandlerRegistration::new(
             external_spec("server/secret"),
-        echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Local,
             None,
             None,
             Capabilities::new().with_api_key("google", "server-secret".to_string()),
-    ));
+        ))
+        .unwrap();
     let discovery_registry = Arc::new(registry);
     let list_handler = services_list_handler(Arc::clone(&discovery_registry));
     let schema_handler = services_schema_handler(Arc::clone(&discovery_registry));
     let mut full = OperationRegistry::new();
     full.register(HandlerRegistration::new(
         external_spec("server/echo"),
-        echo_handler(),
+        HandlerKind::Once(echo_handler()),
         OperationProvenance::Local,
         None,
         None,
         Capabilities::new(),
-    ));
+    ))
+    .unwrap();
     full.register(HandlerRegistration::new(
         external_spec("server/secret"),
-        echo_handler(),
+        HandlerKind::Once(echo_handler()),
         OperationProvenance::Local,
         None,
         None,
         Capabilities::new().with_api_key("google", "server-secret".to_string()),
-    ));
+    ))
+    .unwrap();
     full.register(HandlerRegistration::new(
         services_list_spec(),
-        list_handler,
+        HandlerKind::Once(list_handler),
         OperationProvenance::Local,
         None,
         None,
         Capabilities::new(),
-    ));
+    ))
+    .unwrap();
     full.register(HandlerRegistration::new(
         services_schema_spec(),
-        schema_handler,
+        HandlerKind::Once(schema_handler),
         OperationProvenance::Local,
         None,
         None,
         Capabilities::new(),
-    ));
+    ))
+    .unwrap();
     Arc::new(full)
 }
 
@@ -191,14 +199,16 @@ async fn two_node_call_round_trip() {
     // it as UnknownIssuer since the self-signed cert is not in the platform
     // root store.
     let mut client_registry = OperationRegistry::new();
-    client_registry.register(HandlerRegistration::new(
+    client_registry
+        .register(HandlerRegistration::new(
             external_spec("client/echo"),
-        echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Local,
             None,
             None,
             Capabilities::new(),
-    ));
+        ))
+        .unwrap();
     let client_registry = Arc::new(client_registry);
     let client = CallClient::new(Arc::clone(&client_registry), Arc::new(NoopIdentityProvider));
 

crates/alknet-http/src/adapters/from_mcp/mod.rs

@@ -12,7 +12,9 @@
 use alknet_call::client::{AdapterError, OperationAdapter};
 use alknet_call::protocol::wire::{CallError, ResponseEnvelope};
 use alknet_call::registry::context::OperationContext;
-use alknet_call::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+use alknet_call::registry::registration::{
+    make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+};
 use alknet_call::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
 };
@@ -156,7 +158,7 @@ fn build_registration(
 
     HandlerRegistration::new(
         spec,
-        handler,
+        HandlerKind::Once(handler),
         OperationProvenance::FromMCP,
         None,
         None,

crates/alknet-http/src/adapters/from_openapi.rs

@@ -17,12 +17,16 @@ use std::sync::Arc;
 use alknet_call::client::{AdapterError, OperationAdapter};
 use alknet_call::protocol::wire::{CallError, ResponseEnvelope};
 use alknet_call::registry::context::OperationContext;
-use alknet_call::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+use alknet_call::registry::registration::{
+    make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    ResponseStream,
+};
 use alknet_call::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
 };
 use alknet_core::types::Capabilities;
 use async_trait::async_trait;
+use futures::stream;
 use futures::StreamExt;
 use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE};
 use reqwest::Method;
@@ -438,7 +442,33 @@ impl FromOpenAPI {
             .map(|e| (e.http_status.unwrap_or(0), e.code.clone()))
             .collect();
 
-        let handler = make_handler(move |input: Value, context: OperationContext| {
+        let handler = if op_type == OperationType::Subscription {
+            let stream_handler =
+                make_streaming_handler(move |input: Value, context: OperationContext| {
+                    let path_template = path_template.clone();
+                    let method_upper = method_upper.clone();
+                    let auth_scheme = auth_scheme.clone();
+                    let default_headers = default_headers.clone();
+                    let base_url = base_url.clone();
+                    let namespace = namespace.clone();
+                    let http_client = Arc::clone(&http_client);
+                    let error_status_codes = error_status_codes.clone();
+                    forward_stream(
+                        &http_client,
+                        &base_url,
+                        &path_template,
+                        &method_upper,
+                        &auth_scheme,
+                        &default_headers,
+                        &namespace,
+                        &error_status_codes,
+                        input,
+                        context,
+                    )
+                });
+            HandlerKind::Stream(stream_handler)
+        } else {
+            let once_handler = make_handler(move |input: Value, context: OperationContext| {
                 let path_template = path_template.clone();
                 let method_upper = method_upper.clone();
                 let auth_scheme = auth_scheme.clone();
@@ -465,6 +495,8 @@ impl FromOpenAPI {
                     .await
                 }
             });
+            HandlerKind::Once(once_handler)
+        };
 
         let capabilities = Capabilities::new();
         Ok(HandlerRegistration::new(
@@ -664,10 +696,6 @@ async fn forward(
 
     let status = response.status();
 
-    if op_type == OperationType::Subscription && status.is_success() {
-        return stream_subscription(request_id, response).await;
-    }
-
     if !status.is_success() {
         let code = error_status_codes
             .iter()
@@ -719,35 +747,136 @@ async fn forward(
     }
 }
 
-async fn stream_subscription(request_id: String, response: reqwest::Response) -> ResponseEnvelope {
+#[allow(clippy::too_many_arguments)]
-    let mut stream = response.bytes_stream();
+fn forward_stream(
-    let mut buffer = String::new();
+    http_client: &Arc<SharedHttpClient>,
-    let mut last_event: Option<Value> = None;
+    base_url: &str,
-    while let Some(chunk_result) = stream.next().await {
+    path_template: &str,
-        match chunk_result {
+    method: &str,
-            Ok(chunk) => {
+    auth_scheme: &Option<HttpAuthScheme>,
+    default_headers: &HashMap<String, String>,
+    namespace: &str,
+    error_status_codes: &[(u16, String)],
+    input: Value,
+    context: OperationContext,
+) -> ResponseStream {
+    let request_id = context.request_id.clone();
+
+    let (http_method, url, body, headers) = match build_request(
+        base_url,
+        path_template,
+        method,
+        auth_scheme,
+        default_headers,
+        namespace,
+        &input,
+        &context,
+    ) {
+        Ok(parts) => parts,
+        Err(err) => {
+            return Box::pin(stream::once(async move {
+                ResponseEnvelope::error(request_id, err)
+            }));
+        }
+    };
+
+    let http_client = Arc::clone(http_client);
+    let error_status_codes = error_status_codes.to_vec();
+
+    let request_id_stream = request_id.clone();
+    let error_status_codes_stream = error_status_codes.clone();
+
+    let init = async move {
+        let request_builder = http_client
+            .client()
+            .request(http_method, url.as_str())
+            .headers(headers)
+            .header(ACCEPT, "text/event-stream");
+        let request_builder = match body.as_ref() {
+            Some(b) => {
+                let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null"));
+                request_builder.body(serialized)
+            }
+            None => request_builder,
+        };
+        request_builder.send().await
+    };
+
+    let sse = stream::once(init).flat_map(move |result| {
+        let request_id = request_id_stream.clone();
+        let error_status_codes = error_status_codes_stream.clone();
+        match result {
+            Err(err) => Box::pin(stream::once(async move {
+                ResponseEnvelope::error(
+                    request_id,
+                    CallError::internal(format!("HTTP request failed: {err}")),
+                )
+            })) as ResponseStream,
+            Ok(response) => {
+                let status = response.status();
+                if !status.is_success() {
+                    let code = error_status_codes
+                        .iter()
+                        .find(|(s, _)| *s == status.as_u16())
+                        .map(|(_, c)| c.clone())
+                        .unwrap_or_else(|| format!("HTTP_{}", status.as_u16()));
+                    let message = format!(
+                        "HTTP {}: {}",
+                        status.as_u16(),
+                        status.canonical_reason().unwrap_or("")
+                    );
+                    Box::pin(stream::once(async move {
+                        ResponseEnvelope::error(request_id, CallError::new(code, message, false))
+                    })) as ResponseStream
+                } else {
+                    let request_id_inner = request_id.clone();
+                    Box::pin(
+                        stream::unfold(
+                            (response.bytes_stream(), String::new()),
+                            move |(mut bytes, mut buffer)| {
+                                let request_id = request_id_inner.clone();
+                                async move {
+                                    match bytes.next().await {
+                                        Some(Ok(chunk)) => {
                                             buffer.push_str(&String::from_utf8_lossy(&chunk));
                                             let (events, remaining) = parse_sse_frames(&buffer);
-                buffer = remaining;
+                                            let envelopes: Vec<ResponseEnvelope> = events
-                for event in events {
+                                                .into_iter()
-                    let parsed = if event.data.trim().is_empty() {
+                                                .map(|e| {
+                                                    let parsed = if e.data.trim().is_empty() {
                                                         Value::Null
                                                     } else {
-                        serde_json::from_str(&event.data)
+                                                        serde_json::from_str(&e.data).unwrap_or(
-                            .unwrap_or(Value::String(event.data.clone()))
+                                                            Value::String(e.data.clone()),
+                                                        )
                                                     };
-                    last_event = Some(parsed.clone());
+                                                    ResponseEnvelope::ok(&request_id, parsed)
+                                                })
+                                                .collect();
+                                            Some((envelopes, (bytes, remaining)))
+                                        }
+                                        Some(Err(err)) => {
+                                            let error = CallError::internal(format!(
+                                                "SSE stream error: {err}"
+                                            ));
+                                            Some((
+                                                vec![ResponseEnvelope::error(request_id, error)],
+                                                (bytes, buffer),
+                                            ))
+                                        }
+                                        None => None,
                                     }
                                 }
-            Err(err) => {
+                            },
-                return ResponseEnvelope::error(
+                        )
-                    request_id,
+                        .flat_map(stream::iter),
-                    CallError::internal(format!("SSE stream error: {err}")),
+                    ) as ResponseStream
-                );
                 }
             }
         }
-    ResponseEnvelope::ok(request_id, last_event.unwrap_or(Value::Null))
+    });
+
+    Box::pin(sse)
 }
 
 struct SseEvent {
@@ -1151,7 +1280,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-10", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         assert_eq!(response.request_id, "req-10");
         match response.result {
             Ok(v) => assert_eq!(v, serde_json::json!({"ok":true})),
@@ -1176,7 +1308,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-11", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Err(e) => {
                 assert_eq!(e.code, "HTTP_404");
@@ -1186,6 +1321,34 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn subscription_op_registration_is_handler_kind_stream() {
+        let spec = OpenAPISpec::from_json(
+            r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
+            "paths":{"/stream":{"post":{"operationId":"stream","responses":{"200":{"content":{"text/event-stream":{"schema":{}}}}}}}}}"##,
+        )
+        .unwrap();
+        let bundles = adapter(spec, config("svc", "https://x", None))
+            .import()
+            .await
+            .unwrap();
+        assert!(matches!(bundles[0].handler, HandlerKind::Stream(_)));
+    }
+
+    #[tokio::test]
+    async fn query_op_registration_is_handler_kind_once() {
+        let spec = OpenAPISpec::from_json(
+            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
+            "paths":{"/data":{"get":{"operationId":"data","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#,
+        )
+        .unwrap();
+        let bundles = adapter(spec, config("svc", "https://x", None))
+            .import()
+            .await
+            .unwrap();
+        assert!(matches!(bundles[0].handler, HandlerKind::Once(_)));
+    }
+
     #[tokio::test]
     async fn integration_sse_subscription_streams_responded_events() {
         let sse_body = "data: {\"n\":1}\n\ndata: {\"n\":2}\n\n";
@@ -1201,10 +1364,67 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-12", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let stream = match &registration.handler {
-        assert!(response.result.is_ok());
+            HandlerKind::Stream(h) => h(serde_json::json!({}), ctx),
-        let last = response.result.unwrap();
+            _ => panic!("expected Stream handler"),
-        assert_eq!(last, serde_json::json!({"n":2}));
+        };
+        let collected: Vec<ResponseEnvelope> = stream.collect().await;
+        assert_eq!(collected.len(), 2);
+        assert_eq!(collected[0].result, Ok(serde_json::json!({"n":1})));
+        assert_eq!(collected[1].result, Ok(serde_json::json!({"n":2})));
+        assert_eq!(collected[0].request_id, "req-12");
+        assert_eq!(collected[1].request_id, "req-12");
+    }
+
+    #[tokio::test]
+    async fn integration_sse_subscription_http_error_returns_single_error_envelope() {
+        let base = spawn_echo_server(404, r#"{"error":"missing"}"#, "application/json").await;
+        let spec = OpenAPISpec::from_json(
+            r##"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
+            "paths":{"/stream":{"post":{"operationId":"stream","responses":{
+                "200":{"content":{"text/event-stream":{"schema":{}}}},
+                "404":{"content":{"application/json":{"schema":{"type":"object"}}}}
+            }}}}}"##,
+        )
+        .unwrap();
+        let bundles = adapter(spec, config("svc", &base, None))
+            .import()
+            .await
+            .unwrap();
+        let registration = &bundles[0];
+        let ctx = noop_context("req-err", Capabilities::new());
+        let stream = match &registration.handler {
+            HandlerKind::Stream(h) => h(serde_json::json!({}), ctx),
+            _ => panic!("expected Stream handler"),
+        };
+        let collected: Vec<ResponseEnvelope> = stream.collect().await;
+        assert_eq!(collected.len(), 1);
+        match &collected[0].result {
+            Err(e) => assert_eq!(e.code, "HTTP_404"),
+            other => panic!("expected HTTP_404 error, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn integration_query_forwarding_unchanged_single_response() {
+        let base = spawn_echo_server(200, r#"{"ok":true}"#, "application/json").await;
+        let spec = OpenAPISpec::from_json(
+            r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},
+            "paths":{"/data":{"get":{"operationId":"data","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#,
+        )
+        .unwrap();
+        let bundles = adapter(spec, config("svc", &base, None))
+            .import()
+            .await
+            .unwrap();
+        let registration = &bundles[0];
+        let ctx = noop_context("req-q", Capabilities::new());
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
+        assert_eq!(response.request_id, "req-q");
+        assert_eq!(response.result, Ok(serde_json::json!({"ok":true})));
     }
 
     #[test]
@@ -1447,11 +1667,16 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-16", Capabilities::new());
-        let response = (registration.handler)(
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => {
+                h(
                     serde_json::json!({"id":"42","filter":"new","body":{"name":"widget"}}),
                     ctx,
                 )
-        .await;
+                .await
+            }
+            _ => panic!("expected Once handler"),
+        };
         assert!(
             response.result.is_ok(),
             "expected Ok, got {:?}",
@@ -1483,7 +1708,10 @@ mod tests {
         let registration = &bundles[0];
         let caps = Capabilities::new().with_http_token("openai", "sk-test-token".to_string());
         let ctx = noop_context("req-17", caps);
-        let _ = (registration.handler)(serde_json::json!({}), ctx).await;
+        let _ = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         let captured = rx.await.unwrap();
         assert_eq!(
             captured.headers.get("authorization").unwrap(),
@@ -1519,7 +1747,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-18", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Ok(Value::String(s)) => assert_eq!(s, "hello world"),
             other => panic!("expected String, got {other:?}"),
@@ -1540,7 +1771,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-19", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Err(e) => assert_eq!(e.code, "HTTP_500"),
             other => panic!("expected HTTP_500, got {other:?}"),

crates/alknet-http/src/adapters/to_mcp.rs

@@ -432,7 +432,7 @@ mod tests {
         services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
     };
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance, OperationRegistry,
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
@@ -502,44 +502,52 @@ mod tests {
     ) -> Arc<OperationRegistry> {
         let mut inner = OperationRegistry::new();
         for (name, op_type, acl) in specs {
-            inner.register(HandlerRegistration::new(
+            inner
+                .register(HandlerRegistration::new(
                     external_spec(&name, op_type, acl),
-                make_echo_handler(),
+                    HandlerKind::Once(make_echo_handler()),
                     OperationProvenance::Local,
                     None,
                     None,
                     Capabilities::new(),
-            ));
+                ))
+                .unwrap();
         }
         let inner = Arc::new(inner);
 
         let mut dispatch_registry = OperationRegistry::new();
         for op in inner.list_operations() {
-            dispatch_registry.register(HandlerRegistration::new(
+            dispatch_registry
+                .register(HandlerRegistration::new(
                     external_spec(&op.name, op.op_type, op.access_control.clone()),
-                make_echo_handler(),
+                    HandlerKind::Once(make_echo_handler()),
                     OperationProvenance::Local,
                     None,
                     None,
                     Capabilities::new(),
-            ));
+                ))
+                .unwrap();
         }
-        dispatch_registry.register(HandlerRegistration::new(
+        dispatch_registry
+            .register(HandlerRegistration::new(
                 services_list_spec(),
-            services_list_handler(Arc::clone(&inner)),
+                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
                 OperationProvenance::Local,
                 None,
                 ScopedPeerEnv::empty().into(),
                 Capabilities::new(),
-        ));
+            ))
-        dispatch_registry.register(HandlerRegistration::new(
+            .unwrap();
+        dispatch_registry
+            .register(HandlerRegistration::new(
                 services_schema_spec(),
-            services_schema_handler(Arc::clone(&inner)),
+                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
                 OperationProvenance::Local,
                 None,
                 ScopedPeerEnv::empty().into(),
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(dispatch_registry)
     }
 

crates/alknet-http/src/adapters/to_openapi.rs

@@ -528,7 +528,7 @@ mod tests {
     use super::*;
     use alknet_call::protocol::wire::ResponseEnvelope;
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance,
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::types::Capabilities;
@@ -539,14 +539,16 @@ mod tests {
     }
 
     fn register(registry: &mut OperationRegistry, spec: OperationSpec) {
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 spec,
-            noop_handler(),
+                HandlerKind::Once(noop_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
     }
 
     fn external_spec(name: &str, errors: Vec<ErrorDefinition>) -> OperationSpec {
@@ -1003,7 +1005,8 @@ mod tests {
     #[test]
     fn internal_operations_excluded_from_error_projection() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     "internal/op",
                     OperationType::Query,
@@ -1013,12 +1016,13 @@ mod tests {
                     vec![error("INTERNAL_ERROR", Some(418))],
                     AccessControl::default(),
                 ),
-            noop_handler(),
+                HandlerKind::Once(noop_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let spec = to_openapi(&registry);
         let responses = responses(&spec, PATH_CALL, "post");
         assert!(

crates/alknet-http/src/gateway/dispatch.rs

@@ -26,6 +26,7 @@ use alknet_call::registry::env::LocalOperationEnv;
 use alknet_call::registry::registration::OperationRegistry;
 use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
 use alknet_core::types::Capabilities;
+use futures::stream::BoxStream;
 use serde_json::Value;
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
@@ -70,11 +71,43 @@ impl GatewayDispatch {
         self.registry.invoke(&operation_name, input, context).await
     }
 
+    pub fn invoke_streaming(
+        &self,
+        identity: Option<Identity>,
+        op: &str,
+        input: Value,
+    ) -> BoxStream<'static, ResponseEnvelope> {
+        let operation_name = strip_leading_slash(op).to_string();
+        let request_id = uuid::Uuid::new_v4().to_string();
+        let context = self.build_root_context_streaming(&request_id, &operation_name, identity);
+        self.registry
+            .invoke_streaming(&operation_name, input, context)
+    }
+
     fn build_root_context(
         &self,
         request_id: &str,
         operation_name: &str,
         identity: Option<Identity>,
+    ) -> OperationContext {
+        self.build_root_context_inner(request_id, operation_name, identity, true)
+    }
+
+    fn build_root_context_streaming(
+        &self,
+        request_id: &str,
+        operation_name: &str,
+        identity: Option<Identity>,
+    ) -> OperationContext {
+        self.build_root_context_inner(request_id, operation_name, identity, false)
+    }
+
+    fn build_root_context_inner(
+        &self,
+        request_id: &str,
+        operation_name: &str,
+        identity: Option<Identity>,
+        bounded: bool,
     ) -> OperationContext {
         let registration = self.registry.registration(operation_name);
         let (composition_authority, capabilities, scoped_env) = match registration {
@@ -97,7 +130,7 @@ impl GatewayDispatch {
             forwarded_for: None,
             capabilities,
             metadata: HashMap::new(),
-            deadline: Some(Instant::now() + DEFAULT_TIMEOUT),
+            deadline: bounded.then(|| Instant::now() + DEFAULT_TIMEOUT),
             scoped_env,
             env,
             abort_policy: AbortPolicy::default(),
@@ -117,10 +150,11 @@ mod tests {
         services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
     };
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance,
+        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::AuthToken;
+    use futures::stream::StreamExt;
     use std::sync::Mutex as StdMutex;
 
     struct StaticIdentityProvider {
@@ -187,7 +221,8 @@ mod tests {
 
     fn registry_with(name: &str, visibility: Visibility, acl: AccessControl) -> OperationRegistry {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     name,
                     OperationType::Query,
@@ -197,36 +232,88 @@ mod tests {
                     vec![],
                     acl,
                 ),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         registry
     }
 
     fn registry_with_discovery(inner: Arc<OperationRegistry>) -> OperationRegistry {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
-            services_list_spec(),
-            services_list_handler(Arc::clone(&inner)),
-            OperationProvenance::Local,
-            None,
-            ScopedPeerEnv::empty().into(),
-            Capabilities::new(),
-        ));
-        registry.register(HandlerRegistration::new(
-            services_schema_spec(),
-            services_schema_handler(Arc::clone(&inner)),
-            OperationProvenance::Local,
-            None,
-            ScopedPeerEnv::empty().into(),
-            Capabilities::new(),
-        ));
         registry
+            .register(HandlerRegistration::new(
+                services_list_spec(),
+                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
+                OperationProvenance::Local,
+                None,
+                ScopedPeerEnv::empty().into(),
+                Capabilities::new(),
+            ))
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
+                services_schema_spec(),
+                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
+                OperationProvenance::Local,
+                None,
+                ScopedPeerEnv::empty().into(),
+                Capabilities::new(),
+            ))
+            .unwrap();
+        registry
+    }
+
+    fn subscription_spec(name: &str, visibility: Visibility, acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Subscription,
+            visibility,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+        )
+    }
+
+    fn echo_streaming_handler() -> HandlerKind {
+        HandlerKind::Stream(make_streaming_handler(|input, context| {
+            futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
+        }))
+    }
+
+    fn registry_with_subscription(
+        name: &str,
+        visibility: Visibility,
+        acl: AccessControl,
+    ) -> OperationRegistry {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec(name, visibility, acl),
+                echo_streaming_handler(),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        registry
+    }
+
+    async fn collect_stream(
+        mut stream: BoxStream<'static, ResponseEnvelope>,
+    ) -> Vec<ResponseEnvelope> {
+        let mut out = Vec::new();
+        while let Some(env) = stream.next().await {
+            out.push(env);
+        }
+        out
     }
 
     fn dispatch(
@@ -270,17 +357,20 @@ mod tests {
     #[tokio::test]
     async fn invoke_for_services_list_returns_access_control_filtered_list() {
         let mut inner = OperationRegistry::new();
-        inner.register(HandlerRegistration::new(
+        inner
+            .register(HandlerRegistration::new(
                 external_spec("public/echo", AccessControl::default()),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        inner.register(HandlerRegistration::new(
+            .unwrap();
+        inner
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin/secret",
                     AccessControl {
@@ -288,14 +378,15 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let inner = Arc::new(inner);
         let discovery = Arc::new(registry_with_discovery(Arc::clone(&inner)));
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
@@ -327,16 +418,18 @@ mod tests {
     #[tokio::test]
     async fn invoke_for_services_schema_returns_spec_for_known_op() {
         let mut inner = OperationRegistry::new();
-        inner.register(HandlerRegistration::new(
+        inner
+            .register(HandlerRegistration::new(
                 external_spec("fs/readFile", AccessControl::default()),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let inner = Arc::new(inner);
         let discovery = Arc::new(registry_with_discovery(Arc::clone(&inner)));
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
@@ -373,16 +466,18 @@ mod tests {
     #[tokio::test]
     async fn invoke_for_internal_op_returns_not_found_not_leaked() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret/op", AccessControl::default()),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = dispatch(registry, provider);
@@ -499,16 +594,18 @@ mod tests {
         let caps = Capabilities::new().with_api_key("google", "k".to_string());
 
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("agent/run", AccessControl::default()),
-            make_handler(|input, context| async move {
+                HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
-            }),
+                })),
                 OperationProvenance::Local,
                 Some(authority),
                 Some(scoped.clone()),
                 caps,
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = dispatch(registry, provider);
@@ -532,4 +629,195 @@ mod tests {
         fn assert_concrete<T: Sized>() {}
         assert_concrete::<GatewayDispatch>();
     }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_subscription_returns_handler_stream() {
+        let registry = Arc::new(registry_with_subscription(
+            "events/stream",
+            Visibility::External,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let stream = dp.invoke_streaming(None, "events/stream", serde_json::json!({ "v": 7 }));
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        assert_eq!(items[0].result, Ok(serde_json::json!({ "v": 7 })));
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_strips_leading_slash_from_operation_name() {
+        let registry = Arc::new(registry_with_subscription(
+            "events/stream",
+            Visibility::External,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let stream = dp.invoke_streaming(None, "/events/stream", serde_json::json!({}));
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        assert!(items[0].result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_unknown_op_yields_single_not_found() {
+        let registry = Arc::new(OperationRegistry::new());
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let stream = dp.invoke_streaming(None, "no/such", serde_json::json!({}));
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => {
+                assert_eq!(e.code, "NOT_FOUND");
+                assert!(e.message.contains("no/such"));
+            }
+            other => panic!("expected NOT_FOUND, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_internal_op_from_external_yields_not_found() {
+        let registry = Arc::new(registry_with_subscription(
+            "secret/stream",
+            Visibility::Internal,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let stream = dp.invoke_streaming(None, "secret/stream", serde_json::json!({}));
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => {
+                assert_eq!(e.code, "NOT_FOUND");
+                assert!(e.message.contains("secret/stream"));
+            }
+            other => panic!("expected NOT_FOUND, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_with_none_identity_and_restricted_op_yields_forbidden() {
+        let registry = Arc::new(registry_with_subscription(
+            "admin/stream",
+            Visibility::External,
+            AccessControl {
+                required_scopes: vec!["admin".to_string()],
+                ..Default::default()
+            },
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let stream = dp.invoke_streaming(None, "admin/stream", serde_json::json!({}));
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => {
+                assert_eq!(e.code, "FORBIDDEN");
+                assert_eq!(e.message, "authentication required");
+            }
+            other => panic!("expected FORBIDDEN, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_query_op_yields_invalid_operation_type() {
+        let registry = Arc::new(registry_with(
+            "echo/run",
+            Visibility::External,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let stream = dp.invoke_streaming(None, "echo/run", serde_json::json!({}));
+        let items = collect_stream(stream).await;
+        assert_eq!(items.len(), 1);
+        match &items[0].result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_on_subscription_op_returns_invalid_operation_type() {
+        let registry = Arc::new(registry_with_subscription(
+            "events/stream",
+            Visibility::External,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let response = dp
+            .invoke(None, "events/stream", serde_json::json!({}))
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn build_root_context_streaming_sets_deadline_none() {
+        let registry = Arc::new(registry_with_subscription(
+            "events/stream",
+            Visibility::External,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let ctx = dp.build_root_context_streaming("req-st-1", "events/stream", None);
+        assert!(!ctx.internal, "internal must be false for wire-ingress");
+        assert!(ctx.forwarded_for.is_none(), "forwarded_for must be None");
+        assert!(ctx.parent_request_id.is_none(), "root has no parent");
+        assert!(
+            ctx.deadline.is_none(),
+            "deadline must be None for streaming"
+        );
+    }
+
+    #[test]
+    fn build_root_context_streaming_carries_registration_bundle_fields() {
+        let authority = alknet_call::registry::context::CompositionAuthority::new(
+            "agent",
+            ["fs:read".to_string()],
+        );
+        let scoped = ScopedPeerEnv::new(["fs/readFile"]);
+        let caps = Capabilities::new().with_api_key("google", "k".to_string());
+
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec(
+                    "agent/stream",
+                    Visibility::External,
+                    AccessControl::default(),
+                ),
+                echo_streaming_handler(),
+                OperationProvenance::Local,
+                Some(authority),
+                Some(scoped.clone()),
+                caps,
+            ))
+            .unwrap();
+        let registry = Arc::new(registry);
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = dispatch(registry, provider);
+
+        let ctx = dp.build_root_context_streaming("req-st-2", "agent/stream", None);
+        assert!(ctx.handler_identity.is_some());
+        assert_eq!(ctx.handler_identity.as_ref().unwrap().label, "agent");
+        assert!(ctx.scoped_env.allows("fs/readFile"));
+        assert!(ctx.capabilities.get("google").is_some());
+        assert!(ctx.deadline.is_none());
+    }
 }

crates/alknet-http/src/server/gateway_routes.rs

@@ -295,7 +295,7 @@ mod tests {
         services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
     };
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance,
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType};
     use alknet_core::auth::{AuthToken, Identity};
@@ -376,20 +376,23 @@ mod tests {
 
     fn registry_with_echo() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo/run", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
     fn registry_with_restricted_op() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin/run",
                     AccessControl {
@@ -397,25 +400,28 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
     fn registry_with_internal_op() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret/op"),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -424,37 +430,43 @@ mod tests {
     ) -> Arc<OperationRegistry> {
         let mut inner = OperationRegistry::new();
         for op in inner_ops {
-            inner.register(op);
+            inner.register(op).unwrap();
         }
         let inner = Arc::new(inner);
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 services_list_spec(),
-            services_list_handler(Arc::clone(&inner)),
+                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 services_schema_spec(),
-            services_schema_handler(Arc::clone(&inner)),
+                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         for spec in inner.list_operations() {
             let name = spec.name.clone();
             let reg = inner.registration(&name).unwrap();
-            registry.register(HandlerRegistration::new(
+            registry
+                .register(HandlerRegistration::new(
                     reg.spec.clone(),
-                Arc::clone(&reg.handler),
+                    reg.handler.clone(),
                     reg.provenance,
                     reg.composition_authority.clone(),
                     reg.scoped_env.clone(),
                     reg.capabilities.clone(),
-            ));
+                ))
+                .unwrap();
         }
         Arc::new(registry)
     }
@@ -572,7 +584,7 @@ mod tests {
         let ops = vec![
             HandlerRegistration::new(
                 external_spec("public/echo", AccessControl::default()),
-                echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
@@ -586,7 +598,7 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-                echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
@@ -625,7 +637,7 @@ mod tests {
     async fn schema_returns_full_spec_for_authorized_op() {
         let ops = vec![HandlerRegistration::new(
             external_spec("echo/run", AccessControl::default()),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Local,
             None,
             None,
@@ -657,7 +669,7 @@ mod tests {
                     ..Default::default()
                 },
             ),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Local,
             None,
             None,
@@ -709,22 +721,26 @@ mod tests {
     #[tokio::test]
     async fn batch_internal_op_returns_not_found_in_array() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret/op"),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo/run", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let router = build_router(registry, unused_provider());
         let req = Request::builder()
@@ -823,14 +839,16 @@ mod tests {
     #[test]
     fn is_internal_op_detects_registered_internal_op() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 internal_spec("secret/op"),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         assert!(is_internal_op(&registry, "secret/op"));
         assert!(is_internal_op(&registry, "/secret/op"));
     }
@@ -838,14 +856,16 @@ mod tests {
     #[test]
     fn is_internal_op_false_for_external_op() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo/run", AccessControl::default()),
-            echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         assert!(!is_internal_op(&registry, "echo/run"));
     }
 
@@ -906,7 +926,7 @@ mod tests {
         let ops = vec![
             HandlerRegistration::new(
                 external_spec("public/echo", AccessControl::default()),
-                echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
@@ -920,7 +940,7 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-                echo_handler(),
+                HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
                 None,
                 None,
@@ -953,7 +973,7 @@ mod tests {
     async fn schema_unknown_op_returns_404() {
         let ops = vec![HandlerRegistration::new(
             external_spec("echo/run", AccessControl::default()),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::Local,
             None,
             None,

crates/alknet-http/src/websocket/mod.rs

@@ -18,7 +18,7 @@ mod tests {
     use alknet_call::protocol::wire::{EventEnvelope, ResponseEnvelope, EVENT_RESPONDED};
     use alknet_call::registry::context::AbortPolicy;
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance,
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::{Identity, IdentityProvider};
@@ -77,14 +77,18 @@ mod tests {
 
     fn echo_registry() -> Arc<alknet_call::registry::registration::OperationRegistry> {
         let mut registry = alknet_call::registry::registration::OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo/run"),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+                HandlerKind::Once(make_handler(|input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -174,7 +178,9 @@ mod tests {
         assert!(!env.contains("worker/exec"));
         conn.register_imported(HandlerRegistration::new(
             external_spec("worker/exec"),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
             OperationProvenance::FromCall,
             None,
             None,

crates/alknet-http/src/websocket/overlay.rs

@@ -30,7 +30,7 @@ mod tests {
     };
     use alknet_call::registry::env::{OperationEnv, PeerRef};
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance, OperationRegistry,
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::{Identity, IdentityProvider};
@@ -113,7 +113,9 @@ mod tests {
     ) -> HandlerRegistration {
         HandlerRegistration::new(
             external_spec(name, acl),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
             OperationProvenance::FromCall,
             composition_authority,
             None,
@@ -123,14 +125,18 @@ mod tests {
 
     fn echo_registry() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo/run", AccessControl::default()),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+                HandlerKind::Once(make_handler(|input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -454,9 +460,9 @@ mod tests {
 
         conn.register_imported(HandlerRegistration::new(
             external_spec("ui/dragged", AccessControl::default()),
-            make_handler(|input, ctx| async move {
+            HandlerKind::Once(make_handler(|input, ctx| async move {
                 ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "echoed": input }))
-            }),
+            })),
             OperationProvenance::FromCall,
             None,
             None,
@@ -654,7 +660,7 @@ mod tests {
         };
         conn.register_imported(HandlerRegistration::new(
             subscription_spec("events/stream"),
-            handler,
+            HandlerKind::Once(handler),
             OperationProvenance::FromCall,
             None,
             None,

crates/alknet-http/src/websocket/upgrade.rs

@@ -249,7 +249,7 @@ mod tests {
     };
     use alknet_call::registry::env::OperationEnv;
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance,
+        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::{AuthToken, Identity};
@@ -330,20 +330,25 @@ mod tests {
 
     fn echo_registry() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec("echo/run", AccessControl::default()),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+                HandlerKind::Once(make_handler(|input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
     fn registry_with_restricted_op() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 external_spec(
                     "admin/run",
                     AccessControl {
@@ -351,56 +356,66 @@ mod tests {
                         ..Default::default()
                     },
                 ),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+                HandlerKind::Once(make_handler(|input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
     fn registry_with_subscription() -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
         let count = Arc::new(StdMutex::new(0u32));
-        let handler = make_handler(move |_input, ctx| {
+        let handler = make_streaming_handler(move |_input, ctx| {
             let counter = Arc::clone(&count);
-            async move {
             let mut c = counter.lock().unwrap();
             *c += 1;
             let value = *c;
-                ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "n": value }))
+            futures::stream::iter(vec![ResponseEnvelope::ok(
-            }
+                ctx.request_id,
+                serde_json::json!({ "n": value }),
+            )])
         });
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 subscription_spec("events/stream"),
-            handler,
+                HandlerKind::Stream(handler),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
     fn registry_with_discovery(inner: Arc<OperationRegistry>) -> Arc<OperationRegistry> {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 services_list_spec(),
-            services_list_handler(Arc::clone(&inner)),
+                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
-        registry.register(HandlerRegistration::new(
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
                 services_schema_spec(),
-            services_schema_handler(Arc::clone(&inner)),
+                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         Arc::new(registry)
     }
 
@@ -543,7 +558,8 @@ mod tests {
     #[tokio::test]
     async fn handle_inbound_envelope_internal_op_yields_not_found() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
+        registry
+            .register(HandlerRegistration::new(
                 OperationSpec::new(
                     "secret/op",
                     OperationType::Query,
@@ -553,12 +569,15 @@ mod tests {
                     vec![],
                     AccessControl::default(),
                 ),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+                HandlerKind::Once(make_handler(|input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
                 OperationProvenance::Local,
                 None,
                 None,
                 Capabilities::new(),
-        ));
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = dispatcher(registry, provider);
@@ -753,19 +772,18 @@ mod tests {
         let dp = dispatcher(registry, provider);
         let conn = Arc::new(CallConnection::new_overlay_only(identity("ws-peer")));
 
-        let mut received = Vec::new();
-        for i in 0..3 {
         let request = EventEnvelope::requested(
-                format!("sub-{i}"),
+            "sub-0",
             serde_json::json!({ "operationId": "/events/stream", "input": {} }),
         );
         let out = handle_inbound_envelope(&dp, &conn, request)
             .await
             .expect("response");
-            assert_eq!(out.r#type, EVENT_RESPONDED);
+        assert_eq!(out.r#type, EVENT_ERROR);
-            received.push(out.id);
+        assert_eq!(
-        }
+            out.payload.get("code"),
-        assert_eq!(received.len(), 3);
+            Some(&serde_json::json!("INVALID_OPERATION_TYPE"))
+        );
     }
 
     #[tokio::test]
@@ -868,7 +886,9 @@ mod tests {
 
         conn.register_imported(HandlerRegistration::new(
             external_spec("ui/dragged", AccessControl::default()),
-            make_handler(|input, ctx| async move { ResponseEnvelope::ok(ctx.request_id, input) }),
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
             OperationProvenance::FromCall,
             None,
             None,
@@ -1044,10 +1064,8 @@ mod tests {
             drive_ws_session(socket, &dp, &conn).await;
         });
 
-        let mut got = Vec::new();
-        for i in 0..3 {
         let request = EventEnvelope::requested(
-                format!("sub-ws-{i}"),
+            "sub-ws-0",
             serde_json::json!({ "operationId": "/events/stream", "input": {} }),
         );
         client
@@ -1058,14 +1076,15 @@ mod tests {
         match msg {
             MockMsg::Binary(bytes) => {
                 let env: EventEnvelope = serde_json::from_slice(&bytes).unwrap();
-                    assert_eq!(env.id, format!("sub-ws-{i}"));
+                assert_eq!(env.id, "sub-ws-0");
-                    assert_eq!(env.r#type, EVENT_RESPONDED);
+                assert_eq!(env.r#type, EVENT_ERROR);
-                    got.push(env.id);
+                assert_eq!(
+                    env.payload.get("code"),
+                    Some(&serde_json::json!("INVALID_OPERATION_TYPE"))
+                );
             }
             other => panic!("expected binary, got {other:?}"),
         }
-        }
-        assert_eq!(got.len(), 3);
 
         client.close().await;
         server_handle.await.ok();

tasks/call/client/from-call-streaming-forwarding.md

@@ -1,7 +1,7 @@
 ---
 id: call/client/from-call-streaming-forwarding
 name: Implement from_call streaming forwarding handler (Subscription → CallConnection::subscribe → StreamingHandler)
-status: pending
+status: completed
 depends_on: [call/registry/streaming-handler-handlerkind]
 scope: narrow
 risk: medium
@@ -169,4 +169,4 @@ or the pending entry's removal handles it).
 
 ## Summary
 
-> To be filled on completion
+> Branched build_bundles on spec.op_type: Subscription → make_streaming_forwarding_handler (HandlerKind::Stream), Query/Mutation → existing make_forwarding_handler (HandlerKind::Once). Added CallConnection::subscribe_with_payload() mirroring call_with_payload (registers in PendingRequestMap, abort cascade wired). Streaming forwarding handler reuses build_forwarded_payload for forwarded_for + auth_token (ADR-032). composition_authority: None, scoped_env: None for FromCall streaming leaves. Added 7 unit tests covering all branches and forwarding behavior.

tasks/call/registry/invoke-streaming.md

@@ -1,7 +1,7 @@
 ---
 id: call/registry/invoke-streaming
 name: Implement OperationRegistry::invoke_streaming() returning ResponseStream
-status: pending
+status: completed
 depends_on: [call/registry/streaming-handler-handlerkind]
 scope: narrow
 risk: medium
@@ -167,4 +167,4 @@ streams. The error envelope carries the `request_id` from the context.
 
 ## Summary
 
-> To be filled on completion
+> Added OperationRegistry::invoke_streaming() in crates/alknet-call/src/registry/registration.rs — the streaming dispatch path for Subscription operations. Same visibility + ACL checks as invoke() (provably identical security axis), then dispatches the StreamingHandler and returns its ResponseStream. Pre-handler errors (not-found, forbidden, INVALID_OPERATION_TYPE for non-Subscription ops) yield a single error ResponseEnvelope via stream::once, then end. Added 6 unit tests covering all paths (subscription dispatch, unknown op, query op cross-kind error, internal op from external, ACL denied, internal call using handler_identity).

tasks/call/registry/streaming-handler-handlerkind.md

@@ -1,7 +1,7 @@
 ---
 id: call/registry/streaming-handler-handlerkind
 name: Introduce StreamingHandler, HandlerKind, ResponseStream types and migrate HandlerRegistration to HandlerKind
-status: pending
+status: completed
 depends_on: []
 scope: broad
 risk: medium
@@ -253,4 +253,4 @@ need the explicit `HandlerKind::Once(...)` wrap.
 
 ## Summary
 
-> To be filled on completion
+> Introduced StreamingHandler/ResponseStream type aliases and HandlerKind enum (Once|Stream) + make_streaming_handler() helper in registration.rs; added CallError::invalid_operation_type() (sixth protocol code, retryable: false) in wire.rs; flipped HandlerRegistration.handler to HandlerKind and changed new() signature; builder absorbs wrapping (with_local/with_leaf wrap Handler in Once for Query/Mutation, new with_local_streaming/with_leaf_streaming take StreamingHandler and wrap in Stream for Subscription) with kind/op_type mismatch validation; OperationRegistry::register() now returns Result<(), String> with clear mismatch message; invoke() errors on HandlerKind::Stream with INVALID_OPERATION_TYPE; OverlayOperationEnv::invoke_with_policy matches on HandlerKind (Stream -> INVALID_OPERATION_TYPE); migrated all ~95 HandlerRegistration::new() call sites to wrap in HandlerKind::Once(handler); updated two websocket subscription tests to expect INVALID_OPERATION_TYPE; added unit tests for invoke/register validation, make_streaming_handler, and overlay Stream-kind rejection. All verification passes (build, clippy -D warnings, test, fmt --check) for alknet-call + alknet-http.

tasks/http/adapters/from-openapi-sse-streaming.md

@@ -1,7 +1,7 @@
 ---
 id: http/adapters/from-openapi-sse-streaming
 name: Implement from_openapi Subscription forwarding as StreamingHandler (SSE response → BoxStream<ResponseEnvelope>)
-status: pending
+status: completed
 depends_on: [call/registry/streaming-handler-handlerkind]
 scope: narrow
 risk: medium
@@ -240,4 +240,4 @@ HandlerRegistration::new(spec, handler, OperationProvenance::FromOpenAPI, None,
 
 ## Summary
 
-> To be filled on completion
+> Branched build_registration on op_type: Subscription → make_streaming_handler + forward_stream() (HandlerKind::Stream), Query/Mutation → existing make_handler + forward() (HandlerKind::Once). forward_stream() sends Accept: text/event-stream, streams SSE chunks via stream::unfold over response.bytes_stream(), reusing parse_sse_frames; each data: frame → one ResponseEnvelope::ok(), HTTP error → single ResponseEnvelope::error(), SSE end → ResponseStream ends. Removed stream_subscription() collect-all placeholder. Added 4 tests + updated integration test. 234 tests pass.