feat(http/server/subscribe-sse-streaming): wire /subscribe to invoke_streaming and pipe BoxStream to SSE

Replace the one-event placeholder (subscribe_stream_from_envelope +
envelope_to_sse_stream, which called invoke() and wrapped the single
ResponseEnvelope) with the real streaming path: subscribe_handler now
calls GatewayDispatch::invoke_streaming() and pipes the
BoxStream<ResponseEnvelope> to SSE via subscribe_stream_from_envelope_stream
(futures::StreamExt::map). Each Ok(output) becomes a data: frame; each
Err becomes an event:error frame (terminal — stream ends after it);
natural stream end closes the SSE. Internal ops still return a single
NOT_FOUND error event via subscribe_stream_internal_error (kept). Client
disconnect drops the stream via Rust's Drop (abort cascade per ADR-016).

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

@@ -20,7 +20,7 @@ use crate::protocol::connection::CallConnection;
 use crate::protocol::wire::ResponseEnvelope;
 use crate::registry::context::OperationContext;
 use crate::registry::registration::{
-    Handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    Handler, HandlerKind, HandlerRegistration, OperationProvenance, StreamingHandler,
 };
 use crate::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
@@ -123,14 +123,23 @@ fn build_bundles(
             });
         }
 
-        let handler = make_forwarding_handler(
+        let kind = match spec.op_type {
-            Arc::new(op_summary.connection.clone()),
+            OperationType::Subscription => HandlerKind::Stream(make_streaming_forwarding_handler(
-            remote_name,
+                Arc::new(op_summary.connection.clone()),
-            op_summary.credentials_auth_token.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,
-            HandlerKind::Once(handler),
+            kind,
             OperationProvenance::FromCall,
             None,
             None,
@@ -311,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
@@ -325,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,
@@ -359,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
@@ -391,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};
@@ -724,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());
@@ -824,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/protocol/connection.rs

@@ -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,

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

@@ -17,6 +17,7 @@ use std::time::{Duration, Instant};
 
 use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
 use alknet_core::types::StreamError;
+use futures::stream::StreamExt;
 use serde_json::Value;
 use tokio::task::JoinHandle;
 use tracing::{debug, warn};
@@ -30,11 +31,37 @@ use super::wire::{
 use crate::protocol::adapter::SessionOverlaySource;
 use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
 use crate::registry::env::{LocalOperationEnv, OperationEnv, PeerCompositeEnv};
-use crate::registry::registration::OperationRegistry;
+use crate::registry::registration::{OperationRegistry, ResponseStream};
+use crate::registry::spec::OperationType;
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 const SWEEPER_INTERVAL: Duration = Duration::from_secs(10);
 
+/// Outcome of dispatching a `call.requested` event. The dispatcher branches on
+/// the registered operation's `op_type` (ADR-049 §6): `Query`/`Mutation` produce
+/// a single [`ResponseEnvelope`] (`Once`), `Subscription` produces a
+/// [`ResponseStream`] (`Stream`) that `handle_stream` pumps to the wire.
+///
+/// This enum is the branch point the spec describes ("branches on `op_type` in
+/// `handle_stream`"): `dispatch` returns it and `handle_stream` matches on it,
+/// keeping the Once path (one frame, no `call.completed`) and the Stream path
+/// (each envelope → frame, `call.completed` on natural end) visibly distinct.
+pub enum DispatchResult {
+    Once(ResponseEnvelope),
+    Stream(ResponseStream),
+}
+
+impl std::fmt::Debug for DispatchResult {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DispatchResult::Once(env) => f.debug_tuple("Once").field(env).finish(),
+            DispatchResult::Stream(_) => {
+                f.debug_tuple("Stream").field(&"<ResponseStream>").finish()
+            }
+        }
+    }
+}
+
 /// Shared dispatcher for an established `CallConnection`. Constructed by
 /// both `CallAdapter` (accept path) and `CallClient` (connect path) and used
 /// to run the dispatch loop. Holds no per-connection state; the
@@ -166,6 +193,36 @@ impl Dispatcher {
         request_id: String,
         payload: Value,
     ) -> ResponseEnvelope {
+        match self.dispatch(connection, request_id, payload).await {
+            DispatchResult::Once(envelope) => envelope,
+            DispatchResult::Stream(mut stream) => stream.next().await.unwrap_or_else(|| {
+                ResponseEnvelope::error(
+                    String::new(),
+                    CallError::internal(
+                        "dispatch_requested called on a Subscription op; use the streaming path",
+                    ),
+                )
+            }),
+        }
+    }
+
+    /// Dispatch a `call.requested` event, branching on the registered
+    /// operation's `op_type` (ADR-049 §6). `Query`/`Mutation` → `invoke()` →
+    /// [`DispatchResult::Once`]; `Subscription` → `invoke_streaming()` →
+    /// [`DispatchResult::Stream`]. Unknown ops and ACL failures resolve via
+    /// the registry's own envelope/error paths (Once for `invoke`, a single
+    /// error envelope for `invoke_streaming`).
+    ///
+    /// For the streaming branch the root context's deadline is cleared
+    /// (`deadline: None`): subscriptions are long-running and unbounded — the
+    /// 30s request/response deadline does not apply (ADR-049 §6, call-protocol
+    /// Timeouts). The Once branch keeps the deadline from `build_root_context`.
+    pub async fn dispatch(
+        &self,
+        connection: &Arc<CallConnection>,
+        request_id: String,
+        payload: Value,
+    ) -> DispatchResult {
         let operation_id = payload
             .get("operationId")
             .and_then(|v| v.as_str())
@@ -180,7 +237,13 @@ impl Dispatcher {
 
         let input = payload.get("input").cloned().unwrap_or(Value::Null);
 
-        let context = self.build_root_context(
+        let is_subscription = self
+            .registry
+            .registration(&operation_name)
+            .map(|r| r.spec.op_type == OperationType::Subscription)
+            .unwrap_or(false);
+
+        let mut context = self.build_root_context(
             request_id.clone(),
             &operation_name,
             identity,
@@ -188,7 +251,16 @@ impl Dispatcher {
             connection,
         );
 
-        self.registry.invoke(&operation_name, input, context).await
+        if is_subscription {
+            context.deadline = None;
+            let stream = self
+                .registry
+                .invoke_streaming(&operation_name, input, context);
+            DispatchResult::Stream(stream)
+        } else {
+            let envelope = self.registry.invoke(&operation_name, input, context).await;
+            DispatchResult::Once(envelope)
+        }
     }
 
     pub async fn handle_abort(&self, connection: &Arc<CallConnection>, request_id: &str) {
@@ -225,14 +297,20 @@ impl Dispatcher {
                     let request_id = envelope.id.clone();
                     let payload = envelope.payload.clone();
 
-                    let response = self
+                    match self
-                        .dispatch_requested(&connection, request_id.clone(), payload)
+                        .dispatch(&connection, request_id.clone(), payload)
-                        .await;
+                        .await
-
+                    {
-                    let event: EventEnvelope = response.into();
+                        DispatchResult::Once(response) => {
-                    if let Err(err) = writer.write_frame(&event).await {
+                            let event: EventEnvelope = response.into();
-                        warn!(error = %err, "failed to write response frame; closing stream");
+                            if let Err(err) = writer.write_frame(&event).await {
-                        break;
+                                warn!(error = %err, "failed to write response frame; closing stream");
+                                break;
+                            }
+                        }
+                        DispatchResult::Stream(stream) => {
+                            self.pump_stream(&mut writer, &request_id, stream).await;
+                        }
                     }
                 }
                 EVENT_ABORTED => {
@@ -246,6 +324,43 @@ impl Dispatcher {
         }
     }
 
+    /// Pump a subscription's [`ResponseStream`] to the wire: each
+    /// [`ResponseEnvelope`] becomes an [`EventEnvelope`] frame (`call.responded`
+    /// for `Ok`, `call.error` for `Err`). On natural stream end (the stream
+    /// returned `None` without the last item being an `Err`), write a
+    /// `call.completed` frame. An `Err` envelope is terminal — the stream
+    /// ends after it and we do NOT write `call.completed` (ADR-049 §6).
+    ///
+    /// If a frame write fails the pump stops early; the stream is dropped on
+    /// return, releasing the handler's resources via `Drop` (ADR-016). The
+    /// pump is cancellable: it runs inside the `handle_stream` task, so a
+    /// `call.aborted` for this request ID (handled by `handle_abort` on
+    /// another stream) or connection close cancels the task and drops the
+    /// stream.
+    pub(crate) async fn pump_stream<W: tokio::io::AsyncWrite + Unpin>(
+        &self,
+        writer: &mut super::wire::FrameFramedWriter<W>,
+        request_id: &str,
+        mut stream: ResponseStream,
+    ) {
+        let mut last_was_error = false;
+        while let Some(envelope) = stream.next().await {
+            last_was_error = envelope.result.is_err();
+            let event: EventEnvelope = envelope.into();
+            if let Err(err) = writer.write_frame(&event).await {
+                warn!(error = %err, "failed to write streaming frame; closing stream");
+                return;
+            }
+        }
+
+        if !last_was_error {
+            let completed = EventEnvelope::completed(request_id);
+            if let Err(err) = writer.write_frame(&completed).await {
+                warn!(error = %err, "failed to write call.completed");
+            }
+        }
+    }
+
     /// Run the shared dispatch loop over an established `CallConnection`:
     /// spawn the pending-entry sweeper, accept bidirectional streams until the
     /// connection closes, dispatch each stream via `handle_stream`, and fail
@@ -325,9 +440,9 @@ impl Clone for Dispatcher {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::protocol::wire::EVENT_RESPONDED;
+    use crate::protocol::wire::{EVENT_COMPLETED, EVENT_ERROR, EVENT_RESPONDED};
     use crate::registry::registration::{
-        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
@@ -874,4 +989,388 @@ mod tests {
             Some(&serde_json::json!({ "v": 42 }))
         );
     }
+
+    // --- streaming dispatch branch (ADR-049 §6) ---------------------------
+
+    fn subscription_spec(name: &str, acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Subscription,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+        )
+    }
+
+    fn encode_frame(envelope: &EventEnvelope) -> Vec<u8> {
+        let body = serde_json::to_vec(envelope).expect("serialize envelope");
+        let mut buf = (body.len() as u32).to_be_bytes().to_vec();
+        buf.extend_from_slice(&body);
+        buf
+    }
+
+    async fn read_all_frames(
+        reader: &mut (impl tokio::io::AsyncRead + Unpin),
+    ) -> Vec<EventEnvelope> {
+        let mut buf = Vec::new();
+        use tokio::io::AsyncReadExt;
+        let _ = reader.read_to_end(&mut buf).await;
+        let mut frames = Vec::new();
+        let mut cursor = std::io::Cursor::new(buf);
+        loop {
+            let mut len_buf = [0u8; 4];
+            match tokio::io::AsyncReadExt::read_exact(&mut cursor, &mut len_buf).await {
+                Ok(_) => {}
+                Err(_) => break,
+            }
+            let len = u32::from_be_bytes(len_buf) as usize;
+            let mut body = vec![0u8; len];
+            if tokio::io::AsyncReadExt::read_exact(&mut cursor, &mut body)
+                .await
+                .is_err()
+            {
+                break;
+            }
+            let envelope: EventEnvelope =
+                serde_json::from_slice(&body).expect("deserialize written frame");
+            frames.push(envelope);
+        }
+        frames
+    }
+
+    fn registry_with_subscription(
+        name: &str,
+        handler: crate::registry::registration::StreamingHandler,
+    ) -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec(name, AccessControl::default()),
+                HandlerKind::Stream(handler),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
+    #[tokio::test]
+    async fn dispatch_subscription_returns_stream_result() {
+        let handler = make_streaming_handler(|input, ctx| {
+            futures::stream::iter(vec![
+                ResponseEnvelope::ok(ctx.request_id.clone(), input.clone()),
+                ResponseEnvelope::ok(ctx.request_id.clone(), serde_json::json!({"done": true})),
+            ])
+        });
+        let registry = registry_with_subscription("events/stream", handler);
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let payload = serde_json::json!({
+            "operationId": "/events/stream",
+            "input": { "v": 1 },
+        });
+        match dp.dispatch(&conn, "sub-1".to_string(), payload).await {
+            DispatchResult::Stream(mut stream) => {
+                use futures::stream::StreamExt;
+                let first = stream.next().await.expect("first envelope");
+                assert_eq!(first.request_id, "sub-1");
+                assert_eq!(first.result, Ok(serde_json::json!({ "v": 1 })));
+                let second = stream.next().await.expect("second envelope");
+                assert_eq!(second.result, Ok(serde_json::json!({ "done": true })));
+                assert!(
+                    stream.next().await.is_none(),
+                    "stream ends after two values"
+                );
+            }
+            other => panic!("expected Stream, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn dispatch_subscription_clears_deadline_to_none() {
+        let handler = make_streaming_handler(|_input, ctx| {
+            let deadline = ctx.deadline;
+            futures::stream::iter(vec![ResponseEnvelope::ok(
+                ctx.request_id.clone(),
+                serde_json::json!({ "deadline_is_none": deadline.is_none() }),
+            )])
+        });
+        let registry = registry_with_subscription("events/stream", handler);
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let payload = serde_json::json!({
+            "operationId": "/events/stream",
+            "input": {},
+        });
+        match dp.dispatch(&conn, "sub-dl".to_string(), payload).await {
+            DispatchResult::Stream(mut stream) => {
+                use futures::stream::StreamExt;
+                let env = stream.next().await.expect("one envelope");
+                let out = env.result.expect("ok");
+                assert_eq!(out["deadline_is_none"], Value::Bool(true));
+            }
+            other => panic!("expected Stream, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn dispatch_query_keeps_deadline_some() {
+        let mut registry = OperationRegistry::new();
+        let handler = make_handler(|_input, ctx| async move {
+            let deadline_is_some = ctx.deadline.is_some();
+            ResponseEnvelope::ok(
+                ctx.request_id.clone(),
+                serde_json::json!({ "deadline_is_some": deadline_is_some }),
+            )
+        });
+        registry
+            .register(HandlerRegistration::new(
+                external_spec("echo/run", AccessControl::default()),
+                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);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let payload = serde_json::json!({
+            "operationId": "/echo/run",
+            "input": {},
+        });
+        match dp.dispatch(&conn, "q-1".to_string(), payload).await {
+            DispatchResult::Once(env) => {
+                let out = env.result.expect("ok");
+                assert_eq!(out["deadline_is_some"], Value::Bool(true));
+            }
+            other => panic!("expected Once, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn handle_stream_subscription_pumps_each_frame_then_completed() {
+        let handler = make_streaming_handler(|input, ctx| {
+            let first = input.clone();
+            let rid = ctx.request_id.clone();
+            futures::stream::iter(vec![
+                ResponseEnvelope::ok(rid.clone(), first),
+                ResponseEnvelope::ok(rid.clone(), serde_json::json!({"n": 2})),
+                ResponseEnvelope::ok(rid, serde_json::json!({"n": 3})),
+            ])
+        });
+        let registry = registry_with_subscription("events/stream", handler);
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let request = EventEnvelope::requested(
+            "sub-pump-1",
+            serde_json::json!({
+                "operationId": "/events/stream",
+                "input": { "n": 1 },
+            }),
+        );
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
+        let (send, mut sink) = tokio::io::duplex(8 * 1024);
+        let send = alknet_core::types::SendStream::from_mock(send);
+        let recv = alknet_core::types::RecvStream::from_mock(recv);
+
+        dp.handle_stream(conn, send, recv).await;
+
+        let frames = read_all_frames(&mut sink).await;
+        assert_eq!(frames.len(), 4, "3 responded + 1 completed");
+        for (i, f) in frames[..3].iter().enumerate() {
+            assert_eq!(f.r#type, EVENT_RESPONDED, "frame {i} is call.responded");
+            assert_eq!(f.id, "sub-pump-1");
+        }
+        assert_eq!(frames[3].r#type, EVENT_COMPLETED);
+        assert_eq!(frames[3].id, "sub-pump-1");
+        assert_eq!(frames[3].payload, serde_json::json!({}));
+    }
+
+    #[tokio::test]
+    async fn handle_stream_subscription_error_is_terminal_no_completed() {
+        let handler = make_streaming_handler(|_input, ctx| {
+            let rid = ctx.request_id.clone();
+            futures::stream::iter(vec![
+                ResponseEnvelope::ok(rid.clone(), serde_json::json!({"ok": true})),
+                ResponseEnvelope::error(rid.clone(), CallError::internal("boom")),
+            ])
+        });
+        let registry = registry_with_subscription("events/stream", handler);
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let request = EventEnvelope::requested(
+            "sub-err-1",
+            serde_json::json!({
+                "operationId": "/events/stream",
+                "input": {},
+            }),
+        );
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
+        let (send, mut sink) = tokio::io::duplex(8 * 1024);
+        let send = alknet_core::types::SendStream::from_mock(send);
+        let recv = alknet_core::types::RecvStream::from_mock(recv);
+
+        dp.handle_stream(conn, send, recv).await;
+
+        let frames = read_all_frames(&mut sink).await;
+        assert_eq!(frames.len(), 2, "1 responded + 1 error, no completed");
+        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
+        assert_eq!(frames[1].r#type, EVENT_ERROR);
+        assert_eq!(frames[1].id, "sub-err-1");
+        assert_eq!(
+            frames[1].payload.get("code"),
+            Some(&Value::String("INTERNAL".into()))
+        );
+    }
+
+    #[tokio::test]
+    async fn handle_stream_query_dispatch_unchanged_one_frame_no_completed() {
+        let registry = Arc::new(registry_with(
+            "echo/run",
+            Visibility::External,
+            AccessControl::default(),
+        ));
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let request = EventEnvelope::requested(
+            "q-pump-1",
+            serde_json::json!({
+                "operationId": "/echo/run",
+                "input": { "msg": "hi" },
+            }),
+        );
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
+        let (send, mut sink) = tokio::io::duplex(8 * 1024);
+        let send = alknet_core::types::SendStream::from_mock(send);
+        let recv = alknet_core::types::RecvStream::from_mock(recv);
+
+        dp.handle_stream(conn, send, recv).await;
+
+        let frames = read_all_frames(&mut sink).await;
+        assert_eq!(frames.len(), 1, "query: exactly one frame, no completed");
+        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
+        assert_eq!(frames[0].id, "q-pump-1");
+        assert_eq!(
+            frames[0].payload.get("output"),
+            Some(&serde_json::json!({ "msg": "hi" }))
+        );
+    }
+
+    #[tokio::test]
+    async fn handle_stream_subscription_unknown_op_yields_single_error_no_completed() {
+        let registry = Arc::new(OperationRegistry::new());
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let request = EventEnvelope::requested(
+            "sub-missing-1",
+            serde_json::json!({
+                "operationId": "/no/such/stream",
+                "input": {},
+            }),
+        );
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
+        let (send, mut sink) = tokio::io::duplex(8 * 1024);
+        let send = alknet_core::types::SendStream::from_mock(send);
+        let recv = alknet_core::types::RecvStream::from_mock(recv);
+
+        dp.handle_stream(conn, send, recv).await;
+
+        let frames = read_all_frames(&mut sink).await;
+        assert_eq!(frames.len(), 1, "unknown op: single error, no completed");
+        assert_eq!(frames[0].r#type, EVENT_ERROR);
+        assert_eq!(frames[0].id, "sub-missing-1");
+        assert_eq!(
+            frames[0].payload.get("code"),
+            Some(&Value::String("NOT_FOUND".into()))
+        );
+    }
+
+    #[tokio::test]
+    async fn handle_stream_aborted_for_streaming_request_drops_stream() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc as StdArc;
+
+        let dropped = StdArc::new(AtomicBool::new(false));
+        let dropped_clone = StdArc::clone(&dropped);
+        let handler = make_streaming_handler(move |_input, ctx| {
+            let rid = ctx.request_id.clone();
+            let flag = StdArc::clone(&dropped_clone);
+            struct DropGuard(StdArc<AtomicBool>);
+            impl Drop for DropGuard {
+                fn drop(&mut self) {
+                    self.0.store(true, Ordering::SeqCst);
+                }
+            }
+            let guard = DropGuard(StdArc::clone(&flag));
+            futures::stream::poll_fn(move |_cx| {
+                if flag.load(Ordering::SeqCst) {
+                    return std::task::Poll::Ready(None);
+                }
+                std::task::Poll::Ready(Some(ResponseEnvelope::ok(
+                    rid.clone(),
+                    serde_json::json!({"tick": 1}),
+                )))
+            })
+            .map(move |env| {
+                let _keep_guard = &guard;
+                env
+            })
+        });
+        let registry = registry_with_subscription("events/stream", handler);
+        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
+        let dp = Dispatcher::new(registry, provider);
+        let conn = Arc::new(CallConnection::new(stub_connection()));
+
+        let request = EventEnvelope::requested(
+            "sub-abort-1",
+            serde_json::json!({
+                "operationId": "/events/stream",
+                "input": {},
+            }),
+        );
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
+        let (send, _sink) = tokio::io::duplex(8 * 1024);
+        let send = alknet_core::types::SendStream::from_mock(send);
+        let recv = alknet_core::types::RecvStream::from_mock(recv);
+
+        let conn_clone = Arc::clone(&conn);
+        let dp_clone = dp.clone();
+        let handle = tokio::spawn(async move {
+            dp_clone.handle_stream(conn_clone, send, recv).await;
+        });
+
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        dp.handle_abort(&conn, "sub-abort-1").await;
+        assert!(
+            !conn.pending().lock().contains("sub-abort-1"),
+            "abort removes the pending entry"
+        );
+
+        handle.abort();
+        let _ = handle.await;
+        assert!(
+            dropped.load(Ordering::SeqCst),
+            "stream future dropped → Drop guard released handler resources"
+        );
+    }
 }

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

@@ -4,7 +4,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use alknet_core::types::Capabilities;
-use futures::stream::Stream;
+use futures::stream::{self, Stream};
 use serde_json::Value;
 
 use super::context::{CompositionAuthority, OperationContext, ScopedPeerEnv};
@@ -156,6 +156,63 @@ impl OperationRegistry {
             ),
         }
     }
+
+    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 {
@@ -1006,4 +1063,189 @@ mod tests {
         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-http/src/adapters/from_openapi.rs

@@ -18,13 +18,15 @@ 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, HandlerKind, HandlerRegistration, OperationProvenance,
+    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;
@@ -440,38 +442,66 @@ 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 path_template = path_template.clone();
+            let stream_handler =
-            let method_upper = method_upper.clone();
+                make_streaming_handler(move |input: Value, context: OperationContext| {
-            let auth_scheme = auth_scheme.clone();
+                    let path_template = path_template.clone();
-            let default_headers = default_headers.clone();
+                    let method_upper = method_upper.clone();
-            let base_url = base_url.clone();
+                    let auth_scheme = auth_scheme.clone();
-            let namespace = namespace.clone();
+                    let default_headers = default_headers.clone();
-            let http_client = Arc::clone(&http_client);
+                    let base_url = base_url.clone();
-            let error_status_codes = error_status_codes.clone();
+                    let namespace = namespace.clone();
-            let op_type = op_type;
+                    let http_client = Arc::clone(&http_client);
-            async move {
+                    let error_status_codes = error_status_codes.clone();
-                forward(
+                    forward_stream(
-                    &http_client,
+                        &http_client,
-                    &base_url,
+                        &base_url,
-                    &path_template,
+                        &path_template,
-                    &method_upper,
+                        &method_upper,
-                    &auth_scheme,
+                        &auth_scheme,
-                    &default_headers,
+                        &default_headers,
-                    &namespace,
+                        &namespace,
-                    &error_status_codes,
+                        &error_status_codes,
-                    op_type,
+                        input,
-                    input,
+                        context,
-                    context,
+                    )
-                )
+                });
-                .await
+            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();
+                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();
+                let op_type = op_type;
+                async move {
+                    forward(
+                        &http_client,
+                        &base_url,
+                        &path_template,
+                        &method_upper,
+                        &auth_scheme,
+                        &default_headers,
+                        &namespace,
+                        &error_status_codes,
+                        op_type,
+                        input,
+                        context,
+                    )
+                    .await
+                }
+            });
+            HandlerKind::Once(once_handler)
+        };
 
         let capabilities = Capabilities::new();
         Ok(HandlerRegistration::new(
             spec,
-            HandlerKind::Once(handler),
+            handler,
             OperationProvenance::FromOpenAPI,
             None,
             None,
@@ -666,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()
@@ -721,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>,
-                buffer.push_str(&String::from_utf8_lossy(&chunk));
+    default_headers: &HashMap<String, String>,
-                let (events, remaining) = parse_sse_frames(&buffer);
+    namespace: &str,
-                buffer = remaining;
+    error_status_codes: &[(u16, String)],
-                for event in events {
+    input: Value,
-                    let parsed = if event.data.trim().is_empty() {
+    context: OperationContext,
-                        Value::Null
+) -> ResponseStream {
-                    } else {
+    let request_id = context.request_id.clone();
-                        serde_json::from_str(&event.data)
+
-                            .unwrap_or(Value::String(event.data.clone()))
+    let (http_method, url, body, headers) = match build_request(
-                    };
+        base_url,
-                    last_event = Some(parsed.clone());
+        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);
+                                            let envelopes: Vec<ResponseEnvelope> = events
+                                                .into_iter()
+                                                .map(|e| {
+                                                    let parsed = if e.data.trim().is_empty() {
+                                                        Value::Null
+                                                    } else {
+                                                        serde_json::from_str(&e.data).unwrap_or(
+                                                            Value::String(e.data.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,
+                                    }
+                                }
+                            },
+                        )
+                        .flat_map(stream::iter),
+                    ) as ResponseStream
                 }
             }
-            Err(err) => {
-                return ResponseEnvelope::error(
-                    request_id,
-                    CallError::internal(format!("SSE stream error: {err}")),
-                );
-            }
         }
-    }
+    });
-    ResponseEnvelope::ok(request_id, last_event.unwrap_or(Value::Null))
+
+    Box::pin(sse)
 }
 
 struct SseEvent {
@@ -1194,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";
@@ -1209,13 +1364,67 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-12", 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(), 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!(response.result.is_ok());
+        assert_eq!(response.request_id, "req-q");
-        let last = response.result.unwrap();
+        assert_eq!(response.result, Ok(serde_json::json!({"ok":true})));
-        assert_eq!(last, serde_json::json!({"n":2}));
     }
 
     #[test]

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, HandlerKind, 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 {
@@ -235,6 +269,53 @@ mod tests {
         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(
         registry: Arc<OperationRegistry>,
         provider: Arc<dyn IdentityProvider>,
@@ -548,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

@@ -17,7 +17,8 @@ use axum::response::sse::Event;
 use axum::response::{IntoResponse, Json, Response, Sse};
 use axum::routing::{get, post};
 use axum::Router;
-use futures::stream::{self, BoxStream, Stream};
+use futures::stream::{self, BoxStream};
+use futures::StreamExt;
 use serde::Deserialize;
 use serde_json::{json, Value};
 
@@ -163,18 +164,29 @@ pub(crate) async fn subscribe_handler(
         subscribe_stream_internal_error(request.operation)
     } else {
         let dispatch = state.dispatch();
-        let envelope = dispatch
+        let envelope_stream =
-            .invoke(identity, &request.operation, request.input)
+            dispatch.invoke_streaming(identity, &request.operation, request.input);
-            .await;
+        subscribe_stream_from_envelope_stream(envelope_stream)
-        subscribe_stream_from_envelope(envelope)
     };
     Sse::new(stream)
 }
 
 pub type SubscribeStream = BoxStream<'static, Result<Event, Infallible>>;
 
-fn subscribe_stream_from_envelope(envelope: ResponseEnvelope) -> SubscribeStream {
+fn subscribe_stream_from_envelope_stream(
-    Box::pin(envelope_to_sse_stream(envelope))
+    stream: BoxStream<'static, ResponseEnvelope>,
+) -> SubscribeStream {
+    Box::pin(stream.map(|envelope| match envelope.result {
+        Ok(output) => {
+            let data = serde_json::to_string(&output).unwrap_or_else(|_| "null".to_string());
+            Ok(Event::default().data(data))
+        }
+        Err(error) => {
+            let payload = serde_json::to_value(&error).unwrap_or(Value::Null);
+            let data = serde_json::to_string(&payload).unwrap_or_else(|_| "null".to_string());
+            Ok(Event::default().event("error").data(data))
+        }
+    }))
 }
 
 fn subscribe_stream_internal_error(operation: String) -> SubscribeStream {
@@ -263,24 +275,6 @@ fn is_internal_op(registry: &OperationRegistry, operation: &str) -> bool {
     }
 }
 
-fn envelope_to_sse_stream(
-    envelope: ResponseEnvelope,
-) -> impl Stream<Item = Result<Event, Infallible>> {
-    stream::once(async move {
-        match envelope.result {
-            Ok(output) => {
-                let data = serde_json::to_string(&output).unwrap_or_else(|_| "null".to_string());
-                Ok(Event::default().data(data))
-            }
-            Err(error) => {
-                let payload = serde_json::to_value(&error).unwrap_or(Value::Null);
-                let data = serde_json::to_string(&payload).unwrap_or_else(|_| "null".to_string());
-                Ok(Event::default().event("error").data(data))
-            }
-        }
-    })
-}
-
 fn error_event(operation: &str) -> Result<Event, Infallible> {
     let error = CallError::not_found(operation);
     let payload = serde_json::to_value(&error).unwrap_or(Value::Null);
@@ -295,7 +289,7 @@ mod tests {
         services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
     };
     use alknet_call::registry::registration::{
-        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType};
     use alknet_core::auth::{AuthToken, Identity};
@@ -425,6 +419,73 @@ mod tests {
         Arc::new(registry)
     }
 
+    fn subscription_spec(name: &str, visibility: Visibility, acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Subscription,
+            visibility,
+            json!({}),
+            json!({}),
+            vec![],
+            acl,
+        )
+    }
+
+    fn multi_event_streaming_handler(
+        outputs: Vec<Value>,
+    ) -> alknet_call::registry::registration::StreamingHandler {
+        make_streaming_handler(move |_input, ctx| {
+            let request_id = ctx.request_id.clone();
+            let outputs = outputs.clone();
+            futures::stream::iter(
+                outputs
+                    .into_iter()
+                    .map(move |o| ResponseEnvelope::ok(request_id.clone(), o)),
+            )
+        })
+    }
+
+    fn error_streaming_handler(error: CallError) -> HandlerKind {
+        HandlerKind::Stream(make_streaming_handler(move |_input, ctx| {
+            let request_id = ctx.request_id.clone();
+            let error = error.clone();
+            futures::stream::iter(vec![ResponseEnvelope::error(request_id, error)])
+        }))
+    }
+
+    fn registry_with_subscription_stream(
+        name: &str,
+        outputs: Vec<Value>,
+    ) -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec(name, Visibility::External, AccessControl::default()),
+                HandlerKind::Stream(multi_event_streaming_handler(outputs)),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
+    fn registry_with_subscription_error(name: &str, error: CallError) -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec(name, Visibility::External, AccessControl::default()),
+                error_streaming_handler(error),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
     fn registry_with_discovery_and_ops(
         inner_ops: Vec<HandlerRegistration>,
     ) -> Arc<OperationRegistry> {
@@ -771,15 +832,20 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn subscribe_streams_sse_data_event_until_completed() {
+    async fn subscribe_on_subscription_streams_multiple_data_frames() {
-        let router = build_router(registry_with_echo(), unused_provider());
+        let router = build_router(
+            registry_with_subscription_stream(
+                "events/stream",
+                vec![json!({ "n": 1 }), json!({ "n": 2 }), json!({ "n": 3 })],
+            ),
+            unused_provider(),
+        );
         let req = Request::builder()
             .method("POST")
             .uri("/subscribe")
             .header("content-type", "application/json")
             .body(Body::from(
-                serde_json::to_vec(&json!({ "operation": "echo/run", "input": { "v": 9 } }))
+                serde_json::to_vec(&json!({ "operation": "events/stream", "input": {} })).unwrap(),
-                    .unwrap(),
             ))
             .unwrap();
         let resp = router.oneshot(req).await.unwrap();
@@ -797,10 +863,73 @@ mod tests {
         );
         let bytes = resp.into_body().collect().await.unwrap().to_bytes();
         let body = String::from_utf8_lossy(&bytes);
-        assert!(body.contains("data:"), "expected a data frame, got: {body}");
+        let data_frames = body.matches("data:").count();
+        assert_eq!(data_frames, 3, "expected 3 data frames, got: {body}");
+        assert!(body.contains("\"n\":1"), "expected n=1, got: {body}");
+        assert!(body.contains("\"n\":2"), "expected n=2, got: {body}");
+        assert!(body.contains("\"n\":3"), "expected n=3, got: {body}");
+    }
+
+    #[tokio::test]
+    async fn subscribe_on_subscription_that_yields_error_emits_error_event_then_closes() {
+        let router = build_router(
+            registry_with_subscription_error("events/fail", CallError::internal("handler blew up")),
+            unused_provider(),
+        );
+        let req = Request::builder()
+            .method("POST")
+            .uri("/subscribe")
+            .header("content-type", "application/json")
+            .body(Body::from(
+                serde_json::to_vec(&json!({ "operation": "events/fail", "input": {} })).unwrap(),
+            ))
+            .unwrap();
+        let resp = router.oneshot(req).await.unwrap();
+        assert_eq!(resp.status(), StatusCode::OK);
+        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
+        let body = String::from_utf8_lossy(&bytes);
         assert!(
-            body.contains("\"v\":9"),
+            body.contains("event:error") || body.contains("event: error"),
-            "expected output payload, got: {body}"
+            "expected error event, got: {body}"
+        );
+        assert!(
+            body.contains("INTERNAL"),
+            "expected INTERNAL code, got: {body}"
+        );
+        assert!(
+            body.contains("handler blew up"),
+            "expected error message, got: {body}"
+        );
+        let data_frames = body.matches("data:").count();
+        assert_eq!(
+            data_frames, 1,
+            "expected exactly one data frame (the error payload), got: {body}"
+        );
+    }
+
+    #[tokio::test]
+    async fn subscribe_response_content_type_is_text_event_stream() {
+        let router = build_router(
+            registry_with_subscription_stream("events/stream", vec![json!({ "ok": true })]),
+            unused_provider(),
+        );
+        let req = Request::builder()
+            .method("POST")
+            .uri("/subscribe")
+            .header("content-type", "application/json")
+            .body(Body::from(
+                serde_json::to_vec(&json!({ "operation": "events/stream", "input": {} })).unwrap(),
+            ))
+            .unwrap();
+        let resp = router.oneshot(req).await.unwrap();
+        let ctype = resp
+            .headers()
+            .get(axum::http::header::CONTENT_TYPE)
+            .map(|v| v.to_str().unwrap().to_string());
+        assert_eq!(
+            ctype.as_deref(),
+            Some("text/event-stream"),
+            "expected text/event-stream, got {ctype:?}"
         );
     }
 
@@ -829,6 +958,59 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn subscribe_unknown_op_emits_not_found_error_event() {
+        let router = build_router(
+            registry_with_subscription_stream("events/stream", vec![json!({})]),
+            unused_provider(),
+        );
+        let req = Request::builder()
+            .method("POST")
+            .uri("/subscribe")
+            .header("content-type", "application/json")
+            .body(Body::from(
+                serde_json::to_vec(&json!({ "operation": "no/such", "input": {} })).unwrap(),
+            ))
+            .unwrap();
+        let resp = router.oneshot(req).await.unwrap();
+        assert_eq!(resp.status(), StatusCode::OK);
+        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
+        let body = String::from_utf8_lossy(&bytes);
+        assert!(
+            body.contains("event:error") || body.contains("event: error"),
+            "expected error event, got: {body}"
+        );
+        assert!(
+            body.contains("NOT_FOUND"),
+            "expected NOT_FOUND, got: {body}"
+        );
+    }
+
+    #[tokio::test]
+    async fn subscribe_on_query_op_emits_invalid_operation_type_error_event() {
+        let router = build_router(registry_with_echo(), unused_provider());
+        let req = Request::builder()
+            .method("POST")
+            .uri("/subscribe")
+            .header("content-type", "application/json")
+            .body(Body::from(
+                serde_json::to_vec(&json!({ "operation": "echo/run", "input": {} })).unwrap(),
+            ))
+            .unwrap();
+        let resp = router.oneshot(req).await.unwrap();
+        assert_eq!(resp.status(), StatusCode::OK);
+        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
+        let body = String::from_utf8_lossy(&bytes);
+        assert!(
+            body.contains("event:error") || body.contains("event: error"),
+            "expected error event, got: {body}"
+        );
+        assert!(
+            body.contains("INVALID_OPERATION_TYPE"),
+            "expected INVALID_OPERATION_TYPE, got: {body}"
+        );
+    }
+
     #[test]
     fn is_internal_op_returns_false_for_unknown() {
         let registry = OperationRegistry::new();

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/protocol/dispatch-streaming-branch.md

@@ -1,7 +1,7 @@
 ---
 id: call/protocol/dispatch-streaming-branch
 name: Wire Dispatcher::handle_stream streaming branch (Subscription → invoke_streaming → write each → call.completed)
-status: pending
+status: completed
 depends_on: [call/registry/invoke-streaming]
 scope: narrow
 risk: medium
@@ -171,4 +171,4 @@ task; aborting the connection cancels it).
 
 ## Summary
 
-> To be filled on completion
+> Added DispatchResult enum (Once(ResponseEnvelope) | Stream(ResponseStream)) and Dispatcher::dispatch() branching on op_type (looked up via registry.registration). handle_stream matches on DispatchResult — the branch is visible there (spec framing). Streaming pump writes each ResponseEnvelope → EventEnvelope frame; call.completed on natural end only when !last_was_error (Err is terminal, no call.completed after). deadline: None for streaming branch. Abort via Drop (no new code). Existing Query/Mutation path unchanged. Added 7 unit tests (dispatch branch, deadline clearing, pump frames, error terminal, query unchanged, unknown op, abort drops stream). 306 tests pass.

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

tasks/http/gateway/invoke-streaming.md

@@ -1,7 +1,7 @@
 ---
 id: http/gateway/invoke-streaming
 name: Implement GatewayDispatch::invoke_streaming() returning BoxStream<ResponseEnvelope>
-status: pending
+status: completed
 depends_on: [call/registry/invoke-streaming]
 scope: narrow
 risk: medium
@@ -127,4 +127,4 @@ don't duplicate the logic.
 
 ## Summary
 
-> To be filled on completion
+> Added GatewayDispatch::invoke_streaming() returning BoxStream<ResponseEnvelope>. Security axis provably identical to invoke() via shared build_root_context_inner(bounded: bool); extracted build_root_context_streaming for deadline: None (unbounded subscriptions). Calls OperationRegistry::invoke_streaming(). to_mcp untouched. Added 9 unit tests (all error paths + streaming dispatch + deadline: None verification). 243 tests pass.