feat(call/registry): add OperationRegistry::invoke_streaming() returning ResponseStream

Streaming dispatch path for Subscription operations — counterpart to
invoke(). Same visibility + ACL checks (internal → handler_identity,
external → identity), then dispatches to the StreamingHandler. Pre-handler
errors (not-found, forbidden, INVALID_OPERATION_TYPE for non-Subscription
ops) yield a single error ResponseEnvelope via stream::once and end the
stream. Adds 6 unit tests covering dispatch, not-found, wrong-kind,
internal-from-external, ACL denied, and internal-call handler_identity ACL.

Refs ADR-049 §3, §5.

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

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

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

@@ -19,7 +19,9 @@ use crate::client::AdapterError;
 use crate::protocol::connection::CallConnection;
 use crate::protocol::wire::ResponseEnvelope;
 use crate::registry::context::OperationContext;
-use crate::registry::registration::{Handler, HandlerRegistration, OperationProvenance};
+use crate::registry::registration::{
+    Handler, HandlerKind, HandlerRegistration, OperationProvenance,
+};
 use crate::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
 };
@@ -128,7 +130,7 @@ fn build_bundles(
         );
         bundles.push(HandlerRegistration::new(
             spec,
-            handler,
+            HandlerKind::Once(handler),
             OperationProvenance::FromCall,
             None,
             None,
@@ -549,7 +551,7 @@ mod tests {
         );
         let reg = HandlerRegistration::new(
             spec,
-            handler,
+            HandlerKind::Once(handler),
             OperationProvenance::FromCall,
             None,
             None,

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

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

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

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

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

@@ -26,7 +26,7 @@ use super::wire::{
 use crate::protocol::wire::ResponseEnvelope;
 use crate::registry::context::{generate_request_id, AbortPolicy, OperationContext, ScopedPeerEnv};
 use crate::registry::env::OperationEnv;
-use crate::registry::registration::{Handler, HandlerRegistration};
+use crate::registry::registration::{HandlerKind, HandlerRegistration};
 use crate::registry::spec::AccessResult;
 
 const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(30);
@@ -307,7 +307,7 @@ impl OperationEnv for OverlayOperationEnv {
             return ResponseEnvelope::not_found(parent.request_id.clone(), &name);
         }
 
-        let handler: Handler;
+        let handler: HandlerKind;
         let composition_authority;
         let scoped_env;
         let access_control;
@@ -316,7 +316,7 @@ impl OperationEnv for OverlayOperationEnv {
             let Some(registration) = overlay.get(&name) else {
                 return ResponseEnvelope::not_found(parent.request_id.clone(), &name);
             };
-            handler = Arc::clone(&registration.handler);
+            handler = registration.handler.clone();
             composition_authority = registration.composition_authority.clone();
             scoped_env = registration
                 .scoped_env
@@ -355,7 +355,15 @@ impl OperationEnv for OverlayOperationEnv {
             internal: true,
         };
 
-        handler(input, context).await
+        match handler {
+            HandlerKind::Once(h) => h(input, context).await,
+            HandlerKind::Stream(_) => ResponseEnvelope::error(
+                parent.request_id.clone(),
+                CallError::invalid_operation_type(
+                    "OperationEnv::invoke() called on a Subscription op; composition is request/response-only",
+                ),
+            ),
+        }
     }
 
     fn contains(&self, name: &str) -> bool {
@@ -421,7 +429,7 @@ impl Stream for SubscriptionStream {
 mod tests {
     use super::*;
     use crate::registry::context::CompositionAuthority;
-    use crate::registry::registration::{make_handler, OperationProvenance};
+    use crate::registry::registration::{make_handler, Handler, HandlerKind, OperationProvenance};
     use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::types::{Capabilities, MockConnection};
     use std::collections::HashMap;
@@ -476,7 +484,7 @@ mod tests {
     fn imported_registration(name: &str) -> HandlerRegistration {
         HandlerRegistration::new(
             external_spec(name),
-            echo_handler(),
+            HandlerKind::Once(echo_handler()),
             OperationProvenance::FromCall,
             None,
             None,
@@ -608,7 +616,7 @@ mod tests {
         });
         conn.register_imported(HandlerRegistration::new(
             external_spec("worker/exec"),
-            inspect_handler,
+            HandlerKind::Once(inspect_handler),
             OperationProvenance::FromCall,
             None,
             None,
@@ -631,7 +639,9 @@ mod tests {
     fn connection_accessor_returns_underlying_connection() {
         let conn = CallConnection::new(stub_connection());
         assert_eq!(
-            conn.connection().expect("quic connection present").remote_alpn(),
+            conn.connection()
+                .expect("quic connection present")
+                .remote_alpn(),
             b"alknet/call"
         );
     }
@@ -960,4 +970,39 @@ mod tests {
         assert!(conn.connection().is_some(), "QUIC connection present");
         assert!(conn.identity().is_none(), "no identity set yet");
     }
+
+    #[tokio::test]
+    async fn overlay_env_invoke_on_stream_kind_returns_invalid_operation_type() {
+        use crate::registry::registration::make_streaming_handler;
+        let conn = CallConnection::new(stub_connection());
+        let streaming_handler = make_streaming_handler(|input, ctx| {
+            futures::stream::iter(vec![ResponseEnvelope::ok(ctx.request_id, input)])
+        });
+        conn.register_imported(HandlerRegistration::new(
+            OperationSpec::new(
+                "events/stream",
+                OperationType::Subscription,
+                Visibility::External,
+                serde_json::json!({}),
+                serde_json::json!({}),
+                vec![],
+                AccessControl::default(),
+            ),
+            HandlerKind::Stream(streaming_handler),
+            OperationProvenance::FromCall,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        let env = conn.overlay_env();
+        let scoped = ScopedPeerEnv::new(["events/stream"]);
+        let ctx = root_context("root-stream", scoped, env.clone());
+        let response = env
+            .invoke("events", "stream", serde_json::json!({}), &ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
 }

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

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

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

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

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

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

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

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

crates/alknet-call/tests/two_node_call.rs

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

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

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

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

@@ -17,7 +17,9 @@ use std::sync::Arc;
 use alknet_call::client::{AdapterError, OperationAdapter};
 use alknet_call::protocol::wire::{CallError, ResponseEnvelope};
 use alknet_call::registry::context::OperationContext;
-use alknet_call::registry::registration::{make_handler, HandlerRegistration, OperationProvenance};
+use alknet_call::registry::registration::{
+    make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+};
 use alknet_call::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
 };
@@ -469,7 +471,7 @@ impl FromOpenAPI {
         let capabilities = Capabilities::new();
         Ok(HandlerRegistration::new(
             spec,
-            handler,
+            HandlerKind::Once(handler),
             OperationProvenance::FromOpenAPI,
             None,
             None,
@@ -1151,7 +1153,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-10", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         assert_eq!(response.request_id, "req-10");
         match response.result {
             Ok(v) => assert_eq!(v, serde_json::json!({"ok":true})),
@@ -1176,7 +1181,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-11", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Err(e) => {
                 assert_eq!(e.code, "HTTP_404");
@@ -1201,7 +1209,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-12", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         assert!(response.result.is_ok());
         let last = response.result.unwrap();
         assert_eq!(last, serde_json::json!({"n":2}));
@@ -1447,11 +1458,16 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-16", Capabilities::new());
-        let response = (registration.handler)(
-            serde_json::json!({"id":"42","filter":"new","body":{"name":"widget"}}),
-            ctx,
-        )
-        .await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => {
+                h(
+                    serde_json::json!({"id":"42","filter":"new","body":{"name":"widget"}}),
+                    ctx,
+                )
+                .await
+            }
+            _ => panic!("expected Once handler"),
+        };
         assert!(
             response.result.is_ok(),
             "expected Ok, got {:?}",
@@ -1483,7 +1499,10 @@ mod tests {
         let registration = &bundles[0];
         let caps = Capabilities::new().with_http_token("openai", "sk-test-token".to_string());
         let ctx = noop_context("req-17", caps);
-        let _ = (registration.handler)(serde_json::json!({}), ctx).await;
+        let _ = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         let captured = rx.await.unwrap();
         assert_eq!(
             captured.headers.get("authorization").unwrap(),
@@ -1519,7 +1538,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-18", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Ok(Value::String(s)) => assert_eq!(s, "hello world"),
             other => panic!("expected String, got {other:?}"),
@@ -1540,7 +1562,10 @@ mod tests {
             .unwrap();
         let registration = &bundles[0];
         let ctx = noop_context("req-19", Capabilities::new());
-        let response = (registration.handler)(serde_json::json!({}), ctx).await;
+        let response = match &registration.handler {
+            HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
+            _ => panic!("expected Once handler"),
+        };
         match response.result {
             Err(e) => assert_eq!(e.code, "HTTP_500"),
             other => panic!("expected HTTP_500, got {other:?}"),

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

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

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

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

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

@@ -117,7 +117,7 @@ mod tests {
         services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
     };
     use alknet_call::registry::registration::{
-        make_handler, HandlerRegistration, OperationProvenance,
+        make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
     };
     use alknet_call::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
     use alknet_core::auth::AuthToken;
@@ -187,45 +187,51 @@ mod tests {
 
     fn registry_with(name: &str, visibility: Visibility, acl: AccessControl) -> OperationRegistry {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
-            OperationSpec::new(
-                name,
-                OperationType::Query,
-                visibility,
-                serde_json::json!({}),
-                serde_json::json!({}),
-                vec![],
-                acl,
-            ),
-            make_handler(|input, context| async move {
-                ResponseEnvelope::ok(context.request_id, input)
-            }),
-            OperationProvenance::Local,
-            None,
-            None,
-            Capabilities::new(),
-        ));
+        registry
+            .register(HandlerRegistration::new(
+                OperationSpec::new(
+                    name,
+                    OperationType::Query,
+                    visibility,
+                    serde_json::json!({}),
+                    serde_json::json!({}),
+                    vec![],
+                    acl,
+                ),
+                HandlerKind::Once(make_handler(|input, context| async move {
+                    ResponseEnvelope::ok(context.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
         registry
     }
 
     fn registry_with_discovery(inner: Arc<OperationRegistry>) -> OperationRegistry {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
-            services_list_spec(),
-            services_list_handler(Arc::clone(&inner)),
-            OperationProvenance::Local,
-            None,
-            ScopedPeerEnv::empty().into(),
-            Capabilities::new(),
-        ));
-        registry.register(HandlerRegistration::new(
-            services_schema_spec(),
-            services_schema_handler(Arc::clone(&inner)),
-            OperationProvenance::Local,
-            None,
-            ScopedPeerEnv::empty().into(),
-            Capabilities::new(),
-        ));
+        registry
+            .register(HandlerRegistration::new(
+                services_list_spec(),
+                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
+                OperationProvenance::Local,
+                None,
+                ScopedPeerEnv::empty().into(),
+                Capabilities::new(),
+            ))
+            .unwrap();
+        registry
+            .register(HandlerRegistration::new(
+                services_schema_spec(),
+                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
+                OperationProvenance::Local,
+                None,
+                ScopedPeerEnv::empty().into(),
+                Capabilities::new(),
+            ))
+            .unwrap();
         registry
     }
 
@@ -270,32 +276,36 @@ mod tests {
     #[tokio::test]
     async fn invoke_for_services_list_returns_access_control_filtered_list() {
         let mut inner = OperationRegistry::new();
-        inner.register(HandlerRegistration::new(
-            external_spec("public/echo", AccessControl::default()),
-            make_handler(|input, context| async move {
-                ResponseEnvelope::ok(context.request_id, input)
-            }),
-            OperationProvenance::Local,
-            None,
-            None,
-            Capabilities::new(),
-        ));
-        inner.register(HandlerRegistration::new(
-            external_spec(
-                "admin/secret",
-                AccessControl {
-                    required_scopes: vec!["admin".to_string()],
-                    ..Default::default()
-                },
-            ),
-            make_handler(|input, context| async move {
-                ResponseEnvelope::ok(context.request_id, input)
-            }),
-            OperationProvenance::Local,
-            None,
-            None,
-            Capabilities::new(),
-        ));
+        inner
+            .register(HandlerRegistration::new(
+                external_spec("public/echo", AccessControl::default()),
+                HandlerKind::Once(make_handler(|input, context| async move {
+                    ResponseEnvelope::ok(context.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        inner
+            .register(HandlerRegistration::new(
+                external_spec(
+                    "admin/secret",
+                    AccessControl {
+                        required_scopes: vec!["admin".to_string()],
+                        ..Default::default()
+                    },
+                ),
+                HandlerKind::Once(make_handler(|input, context| async move {
+                    ResponseEnvelope::ok(context.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
         let inner = Arc::new(inner);
         let discovery = Arc::new(registry_with_discovery(Arc::clone(&inner)));
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
@@ -327,16 +337,18 @@ mod tests {
     #[tokio::test]
     async fn invoke_for_services_schema_returns_spec_for_known_op() {
         let mut inner = OperationRegistry::new();
-        inner.register(HandlerRegistration::new(
-            external_spec("fs/readFile", AccessControl::default()),
-            make_handler(|input, context| async move {
-                ResponseEnvelope::ok(context.request_id, input)
-            }),
-            OperationProvenance::Local,
-            None,
-            None,
-            Capabilities::new(),
-        ));
+        inner
+            .register(HandlerRegistration::new(
+                external_spec("fs/readFile", AccessControl::default()),
+                HandlerKind::Once(make_handler(|input, context| async move {
+                    ResponseEnvelope::ok(context.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
         let inner = Arc::new(inner);
         let discovery = Arc::new(registry_with_discovery(Arc::clone(&inner)));
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
@@ -373,16 +385,18 @@ mod tests {
     #[tokio::test]
     async fn invoke_for_internal_op_returns_not_found_not_leaked() {
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
-            internal_spec("secret/op", AccessControl::default()),
-            make_handler(|input, context| async move {
-                ResponseEnvelope::ok(context.request_id, input)
-            }),
-            OperationProvenance::Local,
-            None,
-            None,
-            Capabilities::new(),
-        ));
+        registry
+            .register(HandlerRegistration::new(
+                internal_spec("secret/op", AccessControl::default()),
+                HandlerKind::Once(make_handler(|input, context| async move {
+                    ResponseEnvelope::ok(context.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = dispatch(registry, provider);
@@ -499,16 +513,18 @@ mod tests {
         let caps = Capabilities::new().with_api_key("google", "k".to_string());
 
         let mut registry = OperationRegistry::new();
-        registry.register(HandlerRegistration::new(
-            external_spec("agent/run", AccessControl::default()),
-            make_handler(|input, context| async move {
-                ResponseEnvelope::ok(context.request_id, input)
-            }),
-            OperationProvenance::Local,
-            Some(authority),
-            Some(scoped.clone()),
-            caps,
-        ));
+        registry
+            .register(HandlerRegistration::new(
+                external_spec("agent/run", AccessControl::default()),
+                HandlerKind::Once(make_handler(|input, context| async move {
+                    ResponseEnvelope::ok(context.request_id, input)
+                })),
+                OperationProvenance::Local,
+                Some(authority),
+                Some(scoped.clone()),
+                caps,
+            ))
+            .unwrap();
         let registry = Arc::new(registry);
         let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
         let dp = dispatch(registry, provider);

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

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

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

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

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

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

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

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

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

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