feat: add Pub operation type, HandlerKind::Sink, call.published wire event (ADR-046)

The call protocol had Subscription (server→client streaming) but lacked
the directional complement: client→server streaming, where the initiator
produces a stream and the responder's handler consumes it. This gap was
inherited from the @alkdev/pubsub EventEnvelope prior art, which has
subscribe but no wire-level publish.

ADR-046 adds the Pub primitive:
- OperationType::Pub (client→server streaming)
- OperationType::Subscription renamed to Sub (wire: "subscription" → "sub")
- SinkHandler type + HandlerKind::Sink variant
- PublishStream type alias (Stream<Item = Result<Value, CallError>>)
- OperationRegistry::invoke_sink() dispatch path
- call.published wire event (sixth event type, additive)
- OperationSpec.publish_schema (Option<Value>, validates per-chunk input)
- DispatchResult::Sink + SinkDispatch (handler future + chunk channel)
- Dispatcher::pump_sink (feeds call.published chunks from wire to handler)
- CallConnection::publish() / publish_with_payload() client methods
- from_call sink forwarding handler (make_sink_forwarding_handler)
- make_sink_handler() helper

Fan-out/broker (one producer, N consumers, topic matching) is deferred to
the channels session — the call protocol is point-to-point; the broker is
a routing concern that sits above it. The Pub primitive is the
load-bearing piece the broker will compose on.

- 23 new tests (366 total, up from 343)
- clippy clean, fmt clean

Verification:
  cargo test                                    — 366 passed
  cargo clippy --all-targets -- -D warnings      — clean
  cargo fmt --check                              — clean

src/client/from_call.rs

@@ -21,7 +21,7 @@ use crate::protocol::connection::CallConnection;
 use crate::protocol::wire::ResponseEnvelope;
 use crate::registry::context::OperationContext;
 use crate::registry::registration::{
-    Handler, HandlerKind, HandlerRegistration, OperationProvenance, StreamingHandler,
+    Handler, HandlerKind, HandlerRegistration, OperationProvenance, SinkHandler, StreamingHandler,
 };
 use crate::registry::spec::{
     AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
@@ -124,7 +124,11 @@ fn build_bundles(
         }
 
         let kind = match spec.op_type {
-            OperationType::Subscription => HandlerKind::Stream(make_streaming_forwarding_handler(
+            OperationType::Sub => HandlerKind::Stream(make_streaming_forwarding_handler(
+                Arc::new(op_summary.connection.clone()),
+                remote_name,
+            )),
+            OperationType::Pub => HandlerKind::Sink(make_sink_forwarding_handler(
                 Arc::new(op_summary.connection.clone()),
                 remote_name,
             )),
@@ -254,7 +258,8 @@ fn parse_op_type(s: &str) -> Result<OperationType, AdapterError> {
     match s {
         "query" => Ok(OperationType::Query),
         "mutation" => Ok(OperationType::Mutation),
-        "subscription" => Ok(OperationType::Subscription),
+        "sub" => Ok(OperationType::Sub),
+        "pub" => Ok(OperationType::Pub),
         other => Err(AdapterError::SchemaParse {
             message: format!("unknown op_type: {other}"),
         }),
@@ -381,6 +386,36 @@ fn make_streaming_forwarding_handler(
     })
 }
 
+/// Construct a sink forwarding handler for a `FromCall` `Pub` leaf
+/// (ADR-046): on invocation, calls `CallConnection::publish_with_payload()`
+/// and forwards the local `PublishStream` to the remote as
+/// `call.published` events. The remote's single `call.responded` becomes
+/// the handler's `ResponseEnvelope`. No truncation — the full stream is
+/// forwarded end-to-end.
+///
+/// `forwarded_for` is populated from `context.identity` (ADR-032 §3),
+/// exactly as the request/response and streaming forwarding handlers
+/// do — both via `build_forwarded_payload`.
+fn make_sink_forwarding_handler(
+    connection: Arc<CallConnection>,
+    remote_name: String,
+) -> SinkHandler {
+    use crate::registry::registration::make_sink_handler;
+    use futures::stream::StreamExt;
+    make_sink_handler(move |input, context, publish_stream| {
+        let connection = Arc::clone(&connection);
+        let remote_name = remote_name.clone();
+        async move {
+            let payload = build_forwarded_payload(&remote_name, input, &context);
+            let chunks: Vec<Value> = publish_stream
+                .filter_map(|item| async move { item.ok() })
+                .collect()
+                .await;
+            connection.publish_with_payload(payload, chunks).await
+        }
+    })
+}
+
 /// 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
@@ -810,18 +845,18 @@ mod tests {
         assert_eq!(bundles[0].spec.name, "worker/exec");
     }
 
-    // --- ADR-049 §8: streaming forwarding for Subscription ops -------------
+    // --- ADR-021 §8: streaming forwarding for Sub 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 discovered = vec![op_summary_typed("events/stream", "sub", &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_eq!(bundles[0].spec.op_type, OperationType::Sub);
         assert!(
             matches!(bundles[0].handler, HandlerKind::Stream(_)),
-            "Subscription op must register HandlerKind::Stream"
+            "Sub op must register HandlerKind::Stream"
         );
         assert_eq!(bundles[0].provenance, OperationProvenance::FromCall);
         assert!(bundles[0].composition_authority.is_none());
@@ -860,7 +895,7 @@ mod tests {
         let discovered = vec![
             op_summary_typed("fs/readFile", "query", &conn),
             op_summary_typed("fs/writeFile", "mutation", &conn),
-            op_summary_typed("events/stream", "subscription", &conn),
+            op_summary_typed("events/stream", "sub", &conn),
         ];
         let bundles = build_bundles(discovered, &None, &None).expect("bundles");
         assert_eq!(bundles.len(), 3);
@@ -981,7 +1016,7 @@ mod tests {
         let reg = HandlerRegistration::new(
             OperationSpec::new(
                 "events/stream",
-                OperationType::Subscription,
+                OperationType::Sub,
                 Visibility::External,
                 json!({}),
                 json!({}),

src/protocol/connection.rs

@@ -242,6 +242,125 @@ impl CallConnection {
         self.pending.lock().handle_aborted(request_id);
     }
 
+    /// Publish a stream to a remote `Pub` operation (ADR-046). Sends
+    /// `call.requested` (open), then one `call.published` per chunk in
+    /// `chunks`, then `call.completed` (stream end). Returns the
+    /// responder's single `ResponseEnvelope`.
+    pub async fn publish(
+        &self,
+        operation_id: &str,
+        input: Value,
+        chunks: Vec<Value>,
+    ) -> ResponseEnvelope {
+        let payload = serde_json::json!({
+            "operationId": operation_id,
+            "input": input,
+        });
+        self.publish_with_payload(payload, chunks).await
+    }
+
+    /// Publish a stream to a remote `Pub` 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` for the hub forwarding path used by `from_call`'s
+    /// sink forwarding handler. Mirrors
+    /// [`subscribe_with_payload`](Self::subscribe_with_payload).
+    pub async fn publish_with_payload(
+        &self,
+        payload: Value,
+        chunks: Vec<Value>,
+    ) -> ResponseEnvelope {
+        let request_id = generate_request_id();
+
+        let connection = match &self.connection {
+            Some(c) => c,
+            None => {
+                return ResponseEnvelope::error(
+                    request_id,
+                    CallError::internal("no underlying connection (overlay-only)"),
+                );
+            }
+        };
+
+        let stream = match connection.open_bi().await {
+            Ok(s) => s,
+            Err(err) => {
+                return ResponseEnvelope::error(
+                    request_id,
+                    CallError::internal(format!("failed to open stream: {err}")),
+                );
+            }
+        };
+        let (recv, send) = tokio::io::split(stream);
+
+        let receiver = {
+            let mut pending = self.pending.lock();
+            pending.register_call(
+                request_id.clone(),
+                Instant::now() + DEFAULT_CALL_TIMEOUT,
+                None,
+            )
+        };
+
+        if let Err(err) = self.write_request(send, &request_id, payload).await {
+            let call_error = CallError::internal(err);
+            self.pending
+                .lock()
+                .handle_error(&request_id, call_error.clone());
+            return ResponseEnvelope::error(request_id, call_error);
+        }
+
+        let write_result = self.write_publish_chunks(&request_id, chunks).await;
+        if let Err(err) = write_result {
+            let call_error = CallError::internal(err);
+            self.pending
+                .lock()
+                .handle_error(&request_id, call_error.clone());
+            return ResponseEnvelope::error(request_id, call_error);
+        }
+
+        let pending = Arc::clone(&self.pending);
+        tokio::spawn(async move {
+            read_stream_until_closed(recv, &pending).await;
+        });
+
+        match receiver.await {
+            Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value),
+            Ok(Err(error)) => ResponseEnvelope::error(request_id, error),
+            Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")),
+        }
+    }
+
+    async fn write_publish_chunks(
+        &self,
+        request_id: &str,
+        chunks: Vec<Value>,
+    ) -> Result<(), String> {
+        let connection = self
+            .connection
+            .as_ref()
+            .ok_or_else(|| "no underlying connection (overlay-only)".to_string())?;
+        let stream = connection
+            .open_bi()
+            .await
+            .map_err(|e| format!("failed to open stream: {e}"))?;
+        let (_recv, send) = tokio::io::split(stream);
+        let mut writer = FrameFramedWriter::new(send);
+        for chunk in chunks {
+            let envelope = EventEnvelope::published(request_id, chunk);
+            writer
+                .write_frame(&envelope)
+                .await
+                .map_err(|e| format!("failed to write published frame: {e}"))?;
+        }
+        let completed = EventEnvelope::completed(request_id);
+        writer
+            .write_frame(&completed)
+            .await
+            .map_err(|e| format!("failed to write completed frame: {e}"))?;
+        Ok(())
+    }
+
     async fn write_request<W>(
         &self,
         send: W,
@@ -398,7 +517,13 @@ impl OperationEnv for OverlayOperationEnv {
             HandlerKind::Stream(_) => ResponseEnvelope::error(
                 parent.request_id.clone(),
                 CallError::invalid_operation_type(
-                    "OperationEnv::invoke() called on a Subscription op; composition is request/response-only",
+                    "OperationEnv::invoke() called on a Sub op; composition is request/response-only",
+                ),
+            ),
+            HandlerKind::Sink(_) => ResponseEnvelope::error(
+                parent.request_id.clone(),
+                CallError::invalid_operation_type(
+                    "OperationEnv::invoke() called on a Pub op; composition is request/response-only",
                 ),
             ),
         }
@@ -995,7 +1120,7 @@ mod tests {
         conn.register_imported(HandlerRegistration::new(
             OperationSpec::new(
                 "events/stream",
-                OperationType::Subscription,
+                OperationType::Sub,
                 Visibility::External,
                 serde_json::json!({}),
                 serde_json::json!({}),

src/protocol/dispatch.rs

@@ -11,13 +11,17 @@
 //! See `docs/architecture/` for the spec.
 
 use std::collections::HashMap;
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use crate::core::auth::{AuthToken, Identity, IdentityProvider};
 use crate::core::ownership::OwnershipProvider;
 use crate::core::types::StreamError;
+use futures::channel::mpsc;
 use futures::stream::StreamExt;
+use futures::SinkExt;
 use serde_json::Value;
 use tokio::task::JoinHandle;
 use tracing::{debug, warn};
@@ -31,24 +35,49 @@ 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, ResponseStream};
-use crate::registry::spec::OperationType;
+use crate::registry::registration::{
+    extract_json_pointer, HandlerKind, OperationRegistry, PublishStream, ResponseStream,
+};
+use crate::registry::spec::{AccessResult, OperationType, Visibility};
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 const SWEEPER_INTERVAL: Duration = Duration::from_secs(10);
+const PUBLISH_CHANNEL_BUFFER: usize = 64;
 
 /// 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.
+/// the registered operation's `op_type` (ADR-021 §6, ADR-046):
+/// `Query`/`Mutation` produce a single [`ResponseEnvelope`] (`Once`),
+/// `Sub` produces a [`ResponseStream`] (`Stream`) that `handle_stream`
+/// pumps to the wire, `Pub` produces a [`SinkDispatch`] (`Sink`) that
+/// `handle_stream` feeds from 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.
+/// 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`), the Stream path (each envelope → frame,
+/// `call.completed` on natural end), and the Sink path (feed
+/// `call.published` chunks → one `call.responded`) visibly distinct.
 pub enum DispatchResult {
     Once(ResponseEnvelope),
     Stream(ResponseStream),
+    Sink(SinkDispatch),
+}
+
+/// The sink dispatch result for a `Pub` operation (ADR-046). Carries:
+///
+/// - `handler`: the future returned by `invoke_sink()` — the
+///   `SinkHandler` consuming the `PublishStream` and producing a single
+///   `ResponseEnvelope`.
+/// - `chunk_tx`: the mpsc sender for feeding `call.published` chunks from
+///   the wire into the handler's `PublishStream`.
+///
+/// `handle_stream` reads `call.published` events from the wire, sends
+/// each chunk's `input` into `chunk_tx`, and when the stream ends
+/// (`call.completed` or wire read end), drops `chunk_tx` and awaits
+/// `handler` to get the single `ResponseEnvelope` to write to the wire.
+pub struct SinkDispatch {
+    pub handler: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
+    pub chunk_tx: mpsc::Sender<Result<Value, CallError>>,
 }
 
 impl std::fmt::Debug for DispatchResult {
@@ -58,6 +87,7 @@ impl std::fmt::Debug for DispatchResult {
             DispatchResult::Stream(_) => {
                 f.debug_tuple("Stream").field(&"<ResponseStream>").finish()
             }
+            DispatchResult::Sink(_) => f.debug_tuple("Sink").field(&"<SinkDispatch>").finish(),
         }
     }
 }
@@ -207,24 +237,29 @@ impl Dispatcher {
                 ResponseEnvelope::error(
                     String::new(),
                     CallError::internal(
-                        "dispatch_requested called on a Subscription op; use the streaming path",
+                        "dispatch_requested called on a Sub op; use the streaming path",
                     ),
                 )
             }),
+            DispatchResult::Sink(_) => ResponseEnvelope::error(
+                String::new(),
+                CallError::internal("dispatch_requested called on a Pub op; use the sink 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`).
+    /// operation's `op_type` (ADR-021 §6, ADR-046). `Query`/`Mutation` →
+    /// `invoke()` → [`DispatchResult::Once`]; `Sub` →
+    /// `invoke_streaming()` → [`DispatchResult::Stream`]; `Pub` →
+    /// `invoke_sink()` → [`DispatchResult::Sink`]. Unknown ops and ACL
+    /// failures resolve via the registry's own envelope/error paths.
     ///
-    /// 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`.
+    /// For the `Sub` and `Pub` branches the root context's deadline is
+    /// cleared (`deadline: None`): streaming and publish operations are
+    /// long-running and unbounded — the 30s request/response deadline
+    /// does not apply. The Once branch keeps the deadline from
+    /// `build_root_context`.
     pub async fn dispatch(
         &self,
         connection: &Arc<CallConnection>,
@@ -245,11 +280,11 @@ impl Dispatcher {
 
         let input = payload.get("input").cloned().unwrap_or(Value::Null);
 
-        let is_subscription = self
+        let op_type = self
             .registry
             .registration(&operation_name)
-            .map(|r| r.spec.op_type == OperationType::Subscription)
-            .unwrap_or(false);
+            .map(|r| r.spec.op_type)
+            .unwrap_or(OperationType::Query);
 
         let mut context = self.build_root_context(
             request_id.clone(),
@@ -259,15 +294,87 @@ impl Dispatcher {
             connection,
         );
 
-        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)
+        match op_type {
+            OperationType::Query | OperationType::Mutation => {
+                let envelope = self.registry.invoke(&operation_name, input, context).await;
+                DispatchResult::Once(envelope)
+            }
+            OperationType::Sub => {
+                context.deadline = None;
+                let stream = self
+                    .registry
+                    .invoke_streaming(&operation_name, input, context);
+                DispatchResult::Stream(stream)
+            }
+            OperationType::Pub => {
+                context.deadline = None;
+                let registration = match self.registry.registration(&operation_name) {
+                    Some(r) => r,
+                    None => {
+                        return DispatchResult::Once(ResponseEnvelope::not_found(
+                            request_id.clone(),
+                            &operation_name,
+                        ));
+                    }
+                };
+                if registration.spec.visibility == Visibility::Internal && !context.internal {
+                    return DispatchResult::Once(ResponseEnvelope::not_found(
+                        request_id.clone(),
+                        &operation_name,
+                    ));
+                }
+                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()
+                };
+                let resource_id = registration
+                    .spec
+                    .resource_id_path
+                    .as_ref()
+                    .and_then(|path| extract_json_pointer(&input, path));
+                if let AccessResult::Forbidden(message) = acl.check(
+                    identity.as_ref(),
+                    resource_id.as_deref(),
+                    context.ownership.as_deref(),
+                ) {
+                    return DispatchResult::Once(ResponseEnvelope::forbidden(
+                        request_id.clone(),
+                        message,
+                    ));
+                }
+                let sink_handler = match &registration.handler {
+                    HandlerKind::Sink(h) => Arc::clone(h),
+                    HandlerKind::Once(_) => {
+                        return DispatchResult::Once(ResponseEnvelope::error(
+                            request_id.clone(),
+                            CallError::invalid_operation_type(
+                                "invoke_sink() called on a Query/Mutation op; use invoke()",
+                            ),
+                        ));
+                    }
+                    HandlerKind::Stream(_) => {
+                        return DispatchResult::Once(ResponseEnvelope::error(
+                            request_id.clone(),
+                            CallError::invalid_operation_type(
+                                "invoke_sink() called on a Sub op; use invoke_streaming()",
+                            ),
+                        ));
+                    }
+                };
+                let (chunk_tx, chunk_rx) =
+                    mpsc::channel::<Result<Value, CallError>>(PUBLISH_CHANNEL_BUFFER);
+                let publish_stream: PublishStream = Box::pin(chunk_rx);
+                let handler = (sink_handler)(input, context, publish_stream);
+                DispatchResult::Sink(SinkDispatch {
+                    handler: Box::pin(handler),
+                    chunk_tx,
+                })
+            }
         }
     }
 
@@ -323,6 +430,10 @@ impl Dispatcher {
                         DispatchResult::Stream(stream) => {
                             self.pump_stream(&mut writer, &request_id, stream).await;
                         }
+                        DispatchResult::Sink(sink) => {
+                            self.pump_sink(&mut reader, &mut writer, request_id, sink)
+                                .await;
+                        }
                     }
                 }
                 EVENT_ABORTED => {
@@ -373,6 +484,90 @@ impl Dispatcher {
         }
     }
 
+    /// Pump a `Pub` operation's sink to the wire (ADR-046): feed
+    /// `call.published` chunks from the wire into the `SinkDispatch`'s
+    /// `chunk_tx`, and when the publish stream ends (`call.completed`
+    /// initiator→responder, or wire read end), drop `chunk_tx` and await
+    /// the handler future to get the single `ResponseEnvelope`, which
+    /// is written to the wire as `call.responded` (or `call.error`).
+    ///
+    /// The handler runs concurrently — it consumes the
+    /// `PublishStream` while we feed it. `call.aborted` for this request
+    /// ID (handled by `handle_abort` on another stream) or connection
+    /// close cancels the task and drops the handler future, releasing
+    /// the handler's resources via `Drop` (ADR-020).
+    pub(crate) async fn pump_sink<R, W>(
+        &self,
+        reader: &mut super::wire::FrameFramedReader<R>,
+        writer: &mut super::wire::FrameFramedWriter<W>,
+        request_id: String,
+        sink: SinkDispatch,
+    ) where
+        R: tokio::io::AsyncRead + Unpin,
+        W: tokio::io::AsyncWrite + Unpin,
+    {
+        let SinkDispatch {
+            handler,
+            mut chunk_tx,
+        } = sink;
+
+        let feed_fut = async {
+            loop {
+                match reader.read_frame().await {
+                    Ok(envelope) => {
+                        if envelope.id != request_id {
+                            debug!(
+                                event_type = %envelope.r#type,
+                                id = %envelope.id,
+                                "ignoring event with mismatched id during sink pump"
+                            );
+                            continue;
+                        }
+                        match envelope.r#type.as_str() {
+                            "call.published" => {
+                                let chunk = envelope
+                                    .payload
+                                    .get("input")
+                                    .cloned()
+                                    .unwrap_or(Value::Null);
+                                if chunk_tx.send(Ok(chunk)).await.is_err() {
+                                    break;
+                                }
+                            }
+                            "call.completed" => break,
+                            "call.aborted" => {
+                                let _ = chunk_tx
+                                    .send(Err(CallError::internal("publish aborted by initiator")))
+                                    .await;
+                                break;
+                            }
+                            _ => {
+                                debug!(
+                                    event_type = %envelope.r#type,
+                                    "ignoring non-published event during sink pump"
+                                );
+                            }
+                        }
+                    }
+                    Err(super::wire::FrameError::ConnectionClosed) => break,
+                    Err(err) => {
+                        warn!(error = %err, "frame read error during sink pump; ending stream");
+                        break;
+                    }
+                }
+            }
+            drop(chunk_tx);
+        };
+
+        let (feed_result, response) = tokio::join!(feed_fut, handler);
+        let _: () = feed_result;
+
+        let event: EventEnvelope = response.into();
+        if let Err(err) = writer.write_frame(&event).await {
+            warn!(error = %err, "failed to write sink response frame");
+        }
+    }
+
     /// 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
@@ -985,7 +1180,7 @@ mod tests {
     fn subscription_spec(name: &str, acl: AccessControl) -> OperationSpec {
         OperationSpec::new(
             name,
-            OperationType::Subscription,
+            OperationType::Sub,
             Visibility::External,
             serde_json::json!({}),
             serde_json::json!({}),
@@ -1359,4 +1554,169 @@ mod tests {
             "stream future dropped → Drop guard released handler resources"
         );
     }
+
+    // --- ADR-046: Pub/sink dispatch branch --------------------------------
+
+    fn pub_spec(name: &str, acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Pub,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+            None,
+        )
+    }
+
+    fn registry_with_pub(
+        name: &str,
+        handler: crate::registry::registration::SinkHandler,
+    ) -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec(name, AccessControl::default()),
+                HandlerKind::Sink(handler),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
+    fn counting_sink_handler() -> crate::registry::registration::SinkHandler {
+        use crate::registry::registration::make_sink_handler;
+        use futures::stream::StreamExt;
+        make_sink_handler(|_input, ctx, mut stream| async move {
+            let mut count = 0u32;
+            while let Some(item) = stream.next().await {
+                match item {
+                    Ok(_) => count += 1,
+                    Err(_) => break,
+                }
+            }
+            ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "count": count }))
+        })
+    }
+
+    #[tokio::test]
+    async fn dispatch_pub_returns_sink_result() {
+        let registry = registry_with_pub("fs/upload", counting_sink_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": "/fs/upload",
+            "input": { "path": "/x" },
+        });
+        match dp.dispatch(&conn, "pub-1".to_string(), payload).await {
+            DispatchResult::Sink(sink) => {
+                use futures::SinkExt;
+                let mut tx = sink.chunk_tx;
+                tx.send(Ok(serde_json::json!({"chunk": 1}))).await.unwrap();
+                tx.send(Ok(serde_json::json!({"chunk": 2}))).await.unwrap();
+                drop(tx);
+                let response = sink.handler.await;
+                let out = response.result.expect("ok");
+                assert_eq!(out["count"], serde_json::json!(2));
+            }
+            other => panic!("expected Sink, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn dispatch_pub_clears_deadline_to_none() {
+        let registry = registry_with_pub("fs/upload", counting_sink_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": "/fs/upload",
+            "input": {},
+        });
+        match dp.dispatch(&conn, "pub-dl".to_string(), payload).await {
+            DispatchResult::Sink(sink) => {
+                drop(sink.chunk_tx);
+                let _ = sink.handler.await;
+            }
+            other => panic!("expected Sink, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn handle_stream_pub_feeds_published_chunks_and_returns_response() {
+        let registry = registry_with_pub("fs/upload", counting_sink_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(
+            "pub-pump-1",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": { "path": "/x" },
+            }),
+        );
+        let published1 = EventEnvelope::published("pub-pump-1", serde_json::json!({"chunk": 1}));
+        let published2 = EventEnvelope::published("pub-pump-1", serde_json::json!({"chunk": 2}));
+        let completed = EventEnvelope::completed("pub-pump-1");
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&published1));
+        frame_buf.extend_from_slice(&encode_frame(&published2));
+        frame_buf.extend_from_slice(&encode_frame(&completed));
+
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf));
+        let (send, mut sink) = tokio::io::duplex(8 * 1024);
+        let stream = crate::core::types::BiStream::from_joined(recv, send);
+
+        let handle = tokio::spawn(async move {
+            dp.handle_stream(conn, stream).await;
+        });
+
+        let frames = read_all_frames(&mut sink).await;
+        handle.await.unwrap();
+        assert_eq!(frames.len(), 1, "pub: one response frame, no completed");
+        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
+        assert_eq!(frames[0].id, "pub-pump-1");
+        let output = frames[0].payload.get("output").expect("output field");
+        assert_eq!(output["count"], serde_json::json!(2));
+    }
+
+    #[tokio::test]
+    async fn handle_stream_pub_unknown_op_returns_error() {
+        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(
+            "pub-missing-1",
+            serde_json::json!({
+                "operationId": "/no/such/pub",
+                "input": {},
+            }),
+        );
+        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
+        let (send, mut sink) = tokio::io::duplex(8 * 1024);
+        let stream = crate::core::types::BiStream::from_joined(recv, send);
+
+        dp.handle_stream(conn, stream).await;
+
+        let frames = read_all_frames(&mut sink).await;
+        assert_eq!(frames.len(), 1, "unknown pub op: single error");
+        assert_eq!(frames[0].r#type, EVENT_ERROR);
+        assert_eq!(frames[0].id, "pub-missing-1");
+        assert_eq!(
+            frames[0].payload.get("code"),
+            Some(&Value::String("NOT_FOUND".into()))
+        );
+    }
 }

src/protocol/wire.rs

@@ -14,6 +14,7 @@ pub const EVENT_RESPONDED: &str = "call.responded";
 pub const EVENT_COMPLETED: &str = "call.completed";
 pub const EVENT_ABORTED: &str = "call.aborted";
 pub const EVENT_ERROR: &str = "call.error";
+pub const EVENT_PUBLISHED: &str = "call.published";
 
 const LENGTH_PREFIX_BYTES: usize = 4;
 const MAX_FRAME_SIZE: u32 = 64 * 1024 * 1024;
@@ -47,6 +48,13 @@ impl EventEnvelope {
         Self::new(EVENT_COMPLETED, id, serde_json::json!({}))
     }
 
+    /// A `call.published` event — one chunk of the initiator's data stream
+    /// for a `Pub` operation (ADR-046). The `payload` is
+    /// `{ "input": <Value> }` — the published chunk.
+    pub fn published(id: impl Into<String>, chunk: Value) -> Self {
+        Self::new(EVENT_PUBLISHED, id, serde_json::json!({ "input": chunk }))
+    }
+
     pub fn aborted(id: impl Into<String>) -> Self {
         Self::new(EVENT_ABORTED, id, serde_json::json!({}))
     }
@@ -458,6 +466,31 @@ mod tests {
         assert_eq!(event.payload, serde_json::json!({}));
     }
 
+    #[test]
+    fn event_envelope_published_wraps_input() {
+        let event = EventEnvelope::published("pub-1", Value::String("chunk-data".into()));
+        assert_eq!(event.r#type, EVENT_PUBLISHED);
+        assert_eq!(event.id, "pub-1");
+        assert_eq!(
+            event.payload.get("input"),
+            Some(&Value::String("chunk-data".into()))
+        );
+    }
+
+    #[tokio::test]
+    async fn round_trip_published_envelope() {
+        let (client, server) = duplex(8 * 1024);
+        let envelope = EventEnvelope::published("pub-rt", serde_json::json!({"n": 1}));
+
+        let mut writer = FrameFramedWriter::new(client);
+        writer.write_frame(&envelope).await.unwrap();
+        drop(writer);
+
+        let mut reader = FrameFramedReader::new(server);
+        let read = reader.read_frame().await.unwrap();
+        assert_eq!(read, envelope);
+    }
+
     #[test]
     fn event_envelope_responded_wraps_output() {
         let event = EventEnvelope::responded("req-1", Value::Number(42.into()));

src/registry/discovery.rs

@@ -29,7 +29,7 @@ pub fn services_list_spec() -> OperationSpec {
                             "namespace": { "type": "string" },
                             "op_type": {
                                 "type": "string",
-                                "enum": ["query", "mutation", "subscription"]
+                                "enum": ["query", "mutation", "sub", "pub"]
                             }
                         }
                     }
@@ -83,7 +83,7 @@ pub fn services_list_peers_spec() -> OperationSpec {
                                         "namespace": { "type": "string" },
                                         "op_type": {
                                             "type": "string",
-                                            "enum": ["query", "mutation", "subscription"]
+                                            "enum": ["query", "mutation", "sub", "pub"]
                                         }
                                     }
                                 }
@@ -107,7 +107,7 @@ fn operation_spec_schema() -> Value {
             "namespace": { "type": "string" },
             "op_type": {
                 "type": "string",
-                "enum": ["query", "mutation", "subscription"]
+                "enum": ["query", "mutation", "sub", "pub"]
             },
             "visibility": {
                 "type": "string",
@@ -160,7 +160,8 @@ fn op_type_str(op_type: OperationType) -> &'static str {
     match op_type {
         OperationType::Query => "query",
         OperationType::Mutation => "mutation",
-        OperationType::Subscription => "subscription",
+        OperationType::Sub => "sub",
+        OperationType::Pub => "pub",
     }
 }
 
@@ -542,7 +543,7 @@ mod tests {
             .register(HandlerRegistration::new(
                 OperationSpec::new(
                     "events/subscribe",
-                    OperationType::Subscription,
+                    OperationType::Sub,
                     Visibility::External,
                     json!({}),
                     json!({}),
@@ -752,7 +753,8 @@ mod tests {
     fn op_type_str_matches_wire_enum() {
         assert_eq!(op_type_str(OperationType::Query), "query");
         assert_eq!(op_type_str(OperationType::Mutation), "mutation");
-        assert_eq!(op_type_str(OperationType::Subscription), "subscription");
+        assert_eq!(op_type_str(OperationType::Sub), "sub");
+        assert_eq!(op_type_str(OperationType::Pub), "pub");
     }
 
     #[test]

src/registry/registration.rs

@@ -23,12 +23,36 @@ pub type StreamingHandler = Arc<
         + Sync,
 >;
 
+/// Sink handler — `Pub` operations (ADR-046). Receives the initiator's
+/// data stream (each `call.published` chunk's `input` as one `Value`
+/// item) and returns a single `ResponseEnvelope` (the result of
+/// consuming the stream). An `Err` item in the stream (an initiator-side
+/// error) terminates the stream early; the handler may produce its
+/// result from the partial input or propagate the error.
+pub type SinkHandler = Arc<
+    dyn Fn(
+            Value,
+            OperationContext,
+            Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>,
+        ) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
+        + Send
+        + Sync,
+>;
+
 pub type ResponseStream = Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>;
 
+/// The publish stream shape consumed by a `SinkHandler`. Each item is
+/// one published chunk: `Ok(value)` for a `call.published` event,
+/// `Err(call_error)` for an initiator-side `call.error`. Stream end
+/// (the initiator's `call.completed` or write-half close) yields
+/// `None` — the `SinkHandler` sees a terminated stream.
+pub type PublishStream = Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>;
+
 #[derive(Clone)]
 pub enum HandlerKind {
     Once(Handler),
     Stream(StreamingHandler),
+    Sink(SinkHandler),
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -84,11 +108,13 @@ impl OperationRegistry {
     pub fn register(&mut self, registration: HandlerRegistration) -> Result<(), String> {
         let expected = match registration.spec.op_type {
             OperationType::Query | OperationType::Mutation => "Once",
-            OperationType::Subscription => "Stream",
+            OperationType::Sub => "Stream",
+            OperationType::Pub => "Sink",
         };
         let actual = match registration.handler {
             HandlerKind::Once(_) => "Once",
             HandlerKind::Stream(_) => "Stream",
+            HandlerKind::Sink(_) => "Sink",
         };
         if expected != actual {
             return Err(format!(
@@ -161,9 +187,13 @@ impl OperationRegistry {
             HandlerKind::Stream(_) => ResponseEnvelope::error(
                 request_id,
                 CallError::invalid_operation_type(
-                    "invoke() called on a Subscription op; use invoke_streaming()",
+                    "invoke() called on a Sub op; use invoke_streaming()",
                 ),
             ),
+            HandlerKind::Sink(_) => ResponseEnvelope::error(
+                request_id,
+                CallError::invalid_operation_type("invoke() called on a Pub op; use invoke_sink()"),
+            ),
         }
     }
 
@@ -229,10 +259,90 @@ impl OperationRegistry {
                     )
                 }));
             }
+            HandlerKind::Sink(_) => {
+                return Box::pin(stream::once(async move {
+                    ResponseEnvelope::error(
+                        request_id,
+                        CallError::invalid_operation_type(
+                            "invoke_streaming() called on a Pub op; use invoke_sink()",
+                        ),
+                    )
+                }));
+            }
         };
 
         streaming_handler(input, context)
     }
+
+    /// Dispatch a `Pub` operation (ADR-046). The `publish_stream` is the
+    /// initiator's data stream — each `call.published` chunk's `input` as
+    /// one `Ok(Value)` item, or an initiator-side error as
+    /// `Err(CallError)`. Pre-handler errors (not-found, forbidden,
+    /// `INVALID_OPERATION_TYPE` for a non-Pub op) return a single error
+    /// `ResponseEnvelope`.
+    pub async fn invoke_sink(
+        &self,
+        name: &str,
+        input: Value,
+        publish_stream: PublishStream,
+        context: OperationContext,
+    ) -> ResponseEnvelope {
+        let request_id = context.request_id.clone();
+        let registration = match self.operations.get(name) {
+            Some(r) => r,
+            None => return ResponseEnvelope::not_found(request_id, name),
+        };
+
+        if registration.spec.visibility == Visibility::Internal && !context.internal {
+            return ResponseEnvelope::not_found(request_id, name);
+        }
+
+        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()
+        };
+
+        let resource_id = registration
+            .spec
+            .resource_id_path
+            .as_ref()
+            .and_then(|path| extract_json_pointer(&input, path));
+
+        if let AccessResult::Forbidden(message) = acl.check(
+            identity.as_ref(),
+            resource_id.as_deref(),
+            context.ownership.as_deref(),
+        ) {
+            return ResponseEnvelope::forbidden(request_id, message);
+        }
+
+        let sink_handler = match &registration.handler {
+            HandlerKind::Sink(h) => Arc::clone(h),
+            HandlerKind::Once(_) => {
+                return ResponseEnvelope::error(
+                    request_id,
+                    CallError::invalid_operation_type(
+                        "invoke_sink() called on a Query/Mutation op; use invoke()",
+                    ),
+                );
+            }
+            HandlerKind::Stream(_) => {
+                return ResponseEnvelope::error(
+                    request_id,
+                    CallError::invalid_operation_type(
+                        "invoke_sink() called on a Sub op; use invoke_streaming()",
+                    ),
+                );
+            }
+        };
+
+        (sink_handler)(input, context, publish_stream).await
+    }
 }
 
 impl Default for OperationRegistry {
@@ -261,20 +371,42 @@ impl OperationRegistryBuilder {
     fn wrap_once(spec: &OperationSpec, handler: Handler) -> Result<HandlerKind, String> {
         match spec.op_type {
             OperationType::Query | OperationType::Mutation => Ok(HandlerKind::Once(handler)),
-            OperationType::Subscription => Err(format!(
+            OperationType::Sub => Err(format!(
                 "handler kind mismatch: {:?} requires HandlerKind::Stream (got Handler)",
                 spec.op_type
             )),
+            OperationType::Pub => Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::Sink (got Handler)",
+                spec.op_type
+            )),
         }
     }
 
     fn wrap_stream(spec: &OperationSpec, handler: StreamingHandler) -> Result<HandlerKind, String> {
         match spec.op_type {
-            OperationType::Subscription => Ok(HandlerKind::Stream(handler)),
+            OperationType::Sub => Ok(HandlerKind::Stream(handler)),
             OperationType::Query | OperationType::Mutation => Err(format!(
                 "handler kind mismatch: {:?} requires HandlerKind::Once (got StreamingHandler)",
                 spec.op_type
             )),
+            OperationType::Pub => Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::Sink (got StreamingHandler)",
+                spec.op_type
+            )),
+        }
+    }
+
+    fn wrap_sink(spec: &OperationSpec, handler: SinkHandler) -> Result<HandlerKind, String> {
+        match spec.op_type {
+            OperationType::Pub => Ok(HandlerKind::Sink(handler)),
+            OperationType::Query | OperationType::Mutation => Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::Once (got SinkHandler)",
+                spec.op_type
+            )),
+            OperationType::Sub => Err(format!(
+                "handler kind mismatch: {:?} requires HandlerKind::Stream (got SinkHandler)",
+                spec.op_type
+            )),
         }
     }
 
@@ -372,6 +504,53 @@ impl OperationRegistryBuilder {
         self.store(registration)
     }
 
+    pub fn with_local_sink(
+        self,
+        spec: OperationSpec,
+        handler: SinkHandler,
+        composition_authority: Option<CompositionAuthority>,
+        scoped_env: Option<ScopedPeerEnv>,
+        capabilities: Capabilities,
+    ) -> Result<Self, String> {
+        let kind = Self::wrap_sink(&spec, handler)?;
+        let registration = HandlerRegistration::new(
+            spec,
+            kind,
+            OperationProvenance::Local,
+            composition_authority,
+            scoped_env,
+            capabilities,
+        );
+        self.store(registration)
+    }
+
+    pub fn with_leaf_sink(
+        self,
+        spec: OperationSpec,
+        handler: SinkHandler,
+        capabilities: Capabilities,
+    ) -> Result<Self, String> {
+        self.with_leaf_sink_provenance(
+            spec,
+            handler,
+            OperationProvenance::FromOpenAPI,
+            capabilities,
+        )
+    }
+
+    pub fn with_leaf_sink_provenance(
+        self,
+        spec: OperationSpec,
+        handler: SinkHandler,
+        provenance: OperationProvenance,
+        capabilities: Capabilities,
+    ) -> Result<Self, String> {
+        let kind = Self::wrap_sink(&spec, handler)?;
+        let registration =
+            HandlerRegistration::new(spec, kind, provenance, None, None, capabilities);
+        self.store(registration)
+    }
+
     pub fn with(self, registration: HandlerRegistration) -> Result<Self, String> {
         self.store(registration)
     }
@@ -405,6 +584,14 @@ where
     Arc::new(move |input, context| Box::pin(f(input, context)))
 }
 
+pub fn make_sink_handler<F, Fut>(f: F) -> SinkHandler
+where
+    F: Fn(Value, OperationContext, PublishStream) -> Fut + Send + Sync + 'static,
+    Fut: Future<Output = ResponseEnvelope> + Send + 'static,
+{
+    Arc::new(move |input, context, stream| Box::pin(f(input, context, stream)))
+}
+
 /// Extract a string value from `input` at a JSON-pointer-ish path described
 /// by `$.field` or `$.field/sub` (a leading `$` followed by a slash-separated
 /// pointer, the same shape `serde_json::Value::pointer` expects after the
@@ -1021,7 +1208,7 @@ mod tests {
     fn subscription_spec(name: &str) -> OperationSpec {
         OperationSpec::new(
             name,
-            OperationType::Subscription,
+            OperationType::Sub,
             Visibility::External,
             serde_json::json!({}),
             serde_json::json!({}),
@@ -1093,7 +1280,7 @@ mod tests {
         ));
         match result {
             Err(msg) => assert!(
-                msg.contains("Subscription")
+                msg.contains("Sub")
                     && msg.contains("HandlerKind::Stream")
                     && msg.contains("HandlerKind::Once"),
                 "unexpected message: {msg}"
@@ -1308,7 +1495,7 @@ mod tests {
     fn subscription_spec_with_acl(acl: AccessControl) -> OperationSpec {
         OperationSpec::new(
             "events/stream",
-            OperationType::Subscription,
+            OperationType::Sub,
             Visibility::External,
             serde_json::json!({}),
             serde_json::json!({}),
@@ -1321,7 +1508,7 @@ mod tests {
     fn internal_subscription_spec(acl: AccessControl) -> OperationSpec {
         OperationSpec::new(
             "events/stream",
-            OperationType::Subscription,
+            OperationType::Sub,
             Visibility::Internal,
             serde_json::json!({}),
             serde_json::json!({}),
@@ -1631,4 +1818,454 @@ mod tests {
             "no provider wired → static Identity.resources fallback allows"
         );
     }
+
+    // --- ADR-046: Pub/SinkHandler/invoke_sink tests ------------------------
+
+    fn pub_spec(name: &str, acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Pub,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+            None,
+        )
+    }
+
+    fn internal_pub_spec(acl: AccessControl) -> OperationSpec {
+        OperationSpec::new(
+            "fs/upload",
+            OperationType::Pub,
+            Visibility::Internal,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+            None,
+        )
+    }
+
+    fn collect_sink_handler() -> SinkHandler {
+        use futures::stream::StreamExt;
+        make_sink_handler(|_input, ctx, mut stream| async move {
+            let mut count = 0u32;
+            let mut last = Value::Null;
+            while let Some(item) = stream.next().await {
+                match item {
+                    Ok(v) => {
+                        count += 1;
+                        last = v;
+                    }
+                    Err(e) => {
+                        return ResponseEnvelope::error(
+                            ctx.request_id,
+                            CallError::internal(format!("publish error: {e:?}")),
+                        );
+                    }
+                }
+            }
+            ResponseEnvelope::ok(
+                ctx.request_id,
+                serde_json::json!({ "count": count, "last": last }),
+            )
+        })
+    }
+
+    fn make_publish_stream(items: Vec<Value>) -> PublishStream {
+        Box::pin(futures::stream::iter(
+            items.into_iter().map(Ok::<_, CallError>),
+        ))
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_pub_op_collects_stream_and_returns_result() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec("fs/upload", AccessControl::default()),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-pub-1", None, None, false, ScopedPeerEnv::empty());
+        let stream = make_publish_stream(vec![
+            serde_json::json!({"chunk": 1}),
+            serde_json::json!({"chunk": 2}),
+            serde_json::json!({"chunk": 3}),
+        ]);
+        let response = registry
+            .invoke_sink("fs/upload", serde_json::json!({"path": "/x"}), stream, ctx)
+            .await;
+        let out = response.result.expect("ok");
+        assert_eq!(out["count"], serde_json::json!(3));
+        assert_eq!(out["last"], serde_json::json!({"chunk": 3}));
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_empty_stream_returns_zero_count() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec("fs/upload", AccessControl::default()),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-pub-2", None, None, false, ScopedPeerEnv::empty());
+        let stream = make_publish_stream(vec![]);
+        let response = registry
+            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
+            .await;
+        let out = response.result.expect("ok");
+        assert_eq!(out["count"], serde_json::json!(0));
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_unknown_op_returns_not_found() {
+        let registry = OperationRegistry::new();
+        let ctx = root_context("req-pub-3", None, None, false, ScopedPeerEnv::empty());
+        let stream = make_publish_stream(vec![]);
+        let response = registry
+            .invoke_sink("missing", serde_json::json!({}), stream, ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
+            other => panic!("expected NOT_FOUND, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_internal_op_from_external_returns_not_found() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                internal_pub_spec(AccessControl::default()),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-pub-4", None, None, false, ScopedPeerEnv::empty());
+        let stream = make_publish_stream(vec![]);
+        let response = registry
+            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
+            other => panic!("expected NOT_FOUND, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_acl_denied_returns_forbidden() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec(
+                    "fs/upload",
+                    AccessControl {
+                        required_scopes: vec!["fs:write".to_string()],
+                        ..Default::default()
+                    },
+                ),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context(
+            "req-pub-5",
+            Some(identity_with_scopes("user", &["read"])),
+            None,
+            false,
+            ScopedPeerEnv::empty(),
+        );
+        let stream = make_publish_stream(vec![]);
+        let response = registry
+            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
+            .await;
+        match response.result {
+            Err(e) => {
+                assert_eq!(e.code, "FORBIDDEN");
+                assert!(e.message.contains("fs:write"));
+            }
+            other => panic!("expected FORBIDDEN, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_on_pub_op_returns_invalid_operation_type() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec("fs/upload", AccessControl::default()),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-pub-6", None, None, false, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke("fs/upload", serde_json::json!({}), ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_streaming_on_pub_op_returns_invalid_operation_type() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec("fs/upload", AccessControl::default()),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-pub-7", None, None, false, ScopedPeerEnv::empty());
+        let stream = registry.invoke_streaming("fs/upload", 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_sink_on_query_op_returns_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-pub-8", None, None, false, ScopedPeerEnv::empty());
+        let stream = make_publish_stream(vec![]);
+        let response = registry
+            .invoke_sink("echo", serde_json::json!({}), stream, ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_on_sub_op_returns_invalid_operation_type() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                subscription_spec("events/stream"),
+                HandlerKind::Stream(echo_streaming_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context("req-pub-9", None, None, false, ScopedPeerEnv::empty());
+        let stream = make_publish_stream(vec![]);
+        let response = registry
+            .invoke_sink("events/stream", serde_json::json!({}), stream, ctx)
+            .await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
+            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn register_rejects_once_for_pub_spec() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            pub_spec("fs/upload", AccessControl::default()),
+            HandlerKind::Once(echo_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        match result {
+            Err(msg) => assert!(
+                msg.contains("Pub")
+                    && msg.contains("HandlerKind::Sink")
+                    && msg.contains("HandlerKind::Once"),
+                "unexpected message: {msg}"
+            ),
+            other => panic!("expected Err, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn register_rejects_stream_for_pub_spec() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            pub_spec("fs/upload", AccessControl::default()),
+            HandlerKind::Stream(echo_streaming_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        match result {
+            Err(msg) => assert!(
+                msg.contains("Pub")
+                    && msg.contains("HandlerKind::Sink")
+                    && msg.contains("HandlerKind::Stream"),
+                "unexpected message: {msg}"
+            ),
+            other => panic!("expected Err, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn register_rejects_sink_for_query_spec() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            external_spec("echo", AccessControl::default()),
+            HandlerKind::Sink(collect_sink_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        match result {
+            Err(msg) => assert!(
+                msg.contains("Query") || msg.contains("Mutation"),
+                "unexpected message: {msg}"
+            ),
+            other => panic!("expected Err, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn register_rejects_sink_for_sub_spec() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            subscription_spec("events/stream"),
+            HandlerKind::Sink(collect_sink_handler()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        match result {
+            Err(msg) => assert!(
+                msg.contains("Sub")
+                    && msg.contains("HandlerKind::Stream")
+                    && msg.contains("HandlerKind::Sink"),
+                "unexpected message: {msg}"
+            ),
+            other => panic!("expected Err, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_sink_internal_call_uses_handler_identity_for_acl() {
+        let mut registry = OperationRegistry::new();
+        let composing_authority = CompositionAuthority::new("fs-writer", ["fs:write".to_string()]);
+        registry
+            .register(HandlerRegistration::new(
+                internal_pub_spec(AccessControl {
+                    required_scopes: vec!["fs:write".to_string()],
+                    ..Default::default()
+                }),
+                HandlerKind::Sink(collect_sink_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let ctx = root_context(
+            "req-pub-10",
+            Some(identity_with_scopes("user", &["read"])),
+            Some(composing_authority),
+            true,
+            ScopedPeerEnv::empty(),
+        );
+        let stream = make_publish_stream(vec![serde_json::json!({"chunk": 1})]);
+        let response = registry
+            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
+            .await;
+        let out = response.result.expect("ok — handler identity has fs:write");
+        assert_eq!(out["count"], serde_json::json!(1));
+    }
+
+    #[test]
+    fn builder_with_local_sink_sets_provenance_local() {
+        let registry = OperationRegistryBuilder::new()
+            .with_local_sink(
+                pub_spec("fs/upload", AccessControl::default()),
+                collect_sink_handler(),
+                CompositionAuthority::none(),
+                ScopedPeerEnv::empty().into(),
+                Capabilities::new(),
+            )
+            .unwrap()
+            .build();
+        let reg = registry.registration("fs/upload").expect("registered");
+        assert_eq!(reg.provenance, OperationProvenance::Local);
+        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
+        assert!(reg.composition_authority.is_none());
+        assert!(reg.scoped_env.is_some());
+    }
+
+    #[test]
+    fn builder_with_leaf_sink_sets_provenance_and_no_authority() {
+        let registry = OperationRegistryBuilder::new()
+            .with_leaf_sink(
+                pub_spec("fs/upload", AccessControl::default()),
+                collect_sink_handler(),
+                Capabilities::new(),
+            )
+            .unwrap()
+            .build();
+        let reg = registry.registration("fs/upload").expect("registered");
+        assert_eq!(reg.provenance, OperationProvenance::FromOpenAPI);
+        assert!(reg.composition_authority.is_none());
+        assert!(reg.scoped_env.is_none());
+        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
+    }
+
+    #[test]
+    fn builder_with_leaf_sink_provenance_overrides_provenance() {
+        let registry = OperationRegistryBuilder::new()
+            .with_leaf_sink_provenance(
+                pub_spec("fs/upload", AccessControl::default()),
+                collect_sink_handler(),
+                OperationProvenance::FromCall,
+                Capabilities::new(),
+            )
+            .unwrap()
+            .build();
+        let reg = registry.registration("fs/upload").expect("registered");
+        assert_eq!(reg.provenance, OperationProvenance::FromCall);
+        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
+    }
 }

src/registry/spec.rs

@@ -11,7 +11,8 @@ use serde_json::Value;
 pub enum OperationType {
     Query,
     Mutation,
-    Subscription,
+    Sub,
+    Pub,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -148,11 +149,17 @@ pub struct OperationSpec {
     pub access_control: AccessControl,
     /// JSON pointer into the input for the resource ID, when
     /// `access_control.resource_type` is set and the operation targets a
-    /// specific runtime-spawned resource (ADR-050). e.g. `"$.containerId"`
+    /// specific runtime-spawned resource (ADR-011). e.g. `"$.containerId"`
     /// for `docker/container/exec`. Absent for no-specific-resource
     /// operations (the `list` case). `None` for operations with no
     /// `resource_type` or with static resource sets.
     pub resource_id_path: Option<String>,
+    /// Schema for each published chunk's `input` (Pub ops only, ADR-046).
+    /// `None` for Query/Mutation/Sub ops. When set, the dispatch path
+    /// validates each `call.published` event's `payload.input` against
+    /// this schema before yielding it to the `SinkHandler`. When `None`
+    /// (Pub op with no per-chunk validation), chunks are yielded as-is.
+    pub publish_schema: Option<Value>,
 }
 
 impl OperationSpec {
@@ -184,9 +191,18 @@ impl OperationSpec {
             error_schemas,
             access_control,
             resource_id_path,
+            publish_schema: None,
         }
     }
 
+    /// Set the `publish_schema` (Pub ops only, ADR-046). Validates each
+    /// `call.published` chunk's `input`. Builder-style; returns `self`
+    /// for chaining at registration sites.
+    pub fn with_publish_schema(mut self, schema: Value) -> Self {
+        self.publish_schema = Some(schema);
+        self
+    }
+
     pub fn path(&self) -> String {
         format!("/{}", self.name)
     }
@@ -231,7 +247,7 @@ mod tests {
     fn namespace_derived_from_name() {
         let spec = OperationSpec::new(
             "agent/chat",
-            OperationType::Subscription,
+            OperationType::Sub,
             Visibility::External,
             serde_json::json!({}),
             serde_json::json!({}),