fix: Unit 8 — Pub protocol semantics (P-03, P-04, P-07)

The publish path's decided-but-unimplemented behaviors land: per-chunk
schema validation, initiator call.error terminates the stream, and an
early handler return short-circuits the feed. alktype::validation::
build_validator is now used on both dispatch paths (the dependency was
declared and unused); jsonschema was added as a direct dependency
because its Validator type is part of alktype's public API.

P-03 — OperationSpec::publish_schema was declared, had a builder,
defaulted to None, and was never read. It now round-trips through
discovery: spec_to_json emits "publish_schema" when Some,
operation_spec_schema() advertises the property, and rebuild_spec_for
parses it back (null/absent treated as None, so an imported Pub op no
longer loses the schema). Both dispatch paths (pump_sink and
run_loop_single_stream) validate each call.published chunk's input
against the schema before yielding it to the SinkHandler; on
validation failure an Err(CallError::invalid_input(...)) with the
offending chunk in `details` is injected and the feed terminates,
matching the SinkHandler contract that an Err terminates the stream.
The validator is built once at dispatch time and carried in the
SinkDispatch (and, for the single-stream path, in a new InFlightSink
struct alongside chunk_tx); a schema that fails to compile logs a
warn and falls back to no validation. When publish_schema is None,
chunks are yielded as-is (unchanged behavior).

P-04 — pump_sink matched only call.published/call.completed/call.
aborted and dropped everything else into a debug-log ignore branch; a
call.error from the initiator was silently ignored. Both dispatch
paths now match call.error, parse the CallError from the payload,
inject it as Err(call_error) into the sink's chunk_tx, and terminate
the feed — the handler sees the initiator's error, not a synthetic
"aborted" message. A malformed payload falls back to
CallError::internal("publish error from initiator (malformed)").

P-07 — pump_sink used tokio::join!(feed_fut, handler) and only wrote
the response after both completed; an early-returning handler (e.g.
rejects after chunk 1) had its response deferred until the initiator
finished publishing or the next chunk_tx.send failed. Replaced with a
tokio::select! loop over handler.fuse() and reader.read_frame(): when
the handler completes first, the response is captured and the loop
breaks immediately, the feed is short-circuited (chunk_tx dropped,
the handler's PublishStream sees EOF), and the response is written
without waiting for the feed. The feed-wins branch (natural end /
abort / error / read failure) awaits the handler after the loop as
before. This removes the unbounded stall on a path ADR-046 intends
for long-lived streams.

Tests (15 new, 480 total):
- discovery: spec_to_json emits/omits publish_schema; operation_spec_
  schema documents the property (3).
- from_call: rebuild_spec_for parses publish_schema (present/omitted/
  null) and round-trips with spec_to_json (4).
- dispatch (stream-per-request): publish_schema rejects invalid chunk
  + passes valid chunks + no-schema yields as-is (3); initiator
  call.error terminates with the initiator's error + malformed
  fallback (2); early handler return short-circuits a slow feed and
  the response is not deferred (1).
- channels/client (single-stream): publish_schema rejects invalid
  chunk over channel 0; initiator call.error terminates the publish
  over channel 0 (drives run_loop_single_stream directly with crafted
  frames since the public publish() API takes Stream<Item = Value> and
  cannot emit an initiator call.error) (2).

Verification:
- cargo test              → 480 passed, 0 failed (was 465; +15 new tests)
- cargo clippy --all-targets -- -D warnings → clean
- cargo fmt --check       → clean
- cargo doc --no-deps     → 0 warnings

src/protocol/dispatch.rs

@@ -30,7 +30,7 @@ use super::abort::AbortCascade;
 use super::connection::CallConnection;
 use super::wire::{
     CallError, EventEnvelope, FrameFramedReader, FrameFramedWriter, ResponseEnvelope,
-    EVENT_ABORTED, EVENT_COMPLETED, EVENT_PUBLISHED, EVENT_REQUESTED,
+    EVENT_ABORTED, EVENT_COMPLETED, EVENT_ERROR, EVENT_PUBLISHED, EVENT_REQUESTED,
 };
 use crate::protocol::adapter::SessionOverlaySource;
 use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
@@ -68,6 +68,13 @@ pub enum DispatchResult {
 ///   `ResponseEnvelope`.
 /// - `chunk_tx`: the mpsc sender for feeding `call.published` chunks from
 ///   the wire into the handler's `PublishStream`.
+/// - `publish_validator`: the per-chunk validator built from the op's
+///   `publish_schema` (ADR-046 §4), `None` when the op has no
+///   `publish_schema`. Each `call.published` chunk's `input` is
+///   validated against this before being yielded to the handler; a
+///   validation failure injects an `Err(CallError::invalid_input(...))`
+///   that terminates the stream (matching the `SinkHandler` contract:
+///   an `Err` terminates the stream).
 ///
 /// `handle_stream` reads `call.published` events from the wire, sends
 /// each chunk's `input` into `chunk_tx`, and when the stream ends
@@ -76,6 +83,18 @@ pub enum DispatchResult {
 pub struct SinkDispatch {
     pub(crate) handler: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
     pub(crate) chunk_tx: mpsc::Sender<Result<Value, CallError>>,
+    pub(crate) publish_validator: Option<jsonschema::Validator>,
+}
+
+/// An in-flight Pub (Sink) on the single-stream dispatch loop
+/// (`run_loop_single_stream`), tracked by `request_id`. Carries the
+/// feed's `chunk_tx` and the per-chunk validator (built once at
+/// dispatch time from the op's `publish_schema`). `call.published`
+/// frames arriving after the `call.requested` are routed here;
+/// `call.completed` / `call.aborted` / `call.error` remove the entry.
+struct InFlightSink {
+    chunk_tx: mpsc::Sender<Result<Value, CallError>>,
+    publish_validator: Option<jsonschema::Validator>,
 }
 
 impl std::fmt::Debug for DispatchResult {
@@ -315,6 +334,21 @@ impl Dispatcher {
                         Ok(h) => h,
                         Err(envelope) => return DispatchResult::Once(envelope),
                     };
+                let publish_validator = self
+                    .registry
+                    .registration(&operation_name)
+                    .and_then(|r| r.spec.publish_schema.as_ref())
+                    .and_then(|schema| match alktype::validation::build_validator(schema) {
+                        Ok(v) => Some(v),
+                        Err(e) => {
+                            warn!(
+                                operation = %operation_name,
+                                error = %e,
+                                "publish_schema failed to compile; chunks will not be validated",
+                            );
+                            None
+                        }
+                    });
                 let (chunk_tx, chunk_rx) =
                     mpsc::channel::<Result<Value, CallError>>(PUBLISH_CHANNEL_BUFFER);
                 let publish_stream: PublishStream = Box::pin(chunk_rx);
@@ -322,6 +356,7 @@ impl Dispatcher {
                 DispatchResult::Sink(SinkDispatch {
                     handler: Box::pin(handler),
                     chunk_tx,
+                    publish_validator,
                 })
             }
         }
@@ -441,10 +476,32 @@ impl Dispatcher {
     /// is written to the wire as `call.responded` (or `call.error`).
     ///
     /// The handler runs concurrently — it consumes the
-    /// `PublishStream` while we feed it. On `call.aborted` for this
-    /// request ID, the feed injects an `Err` into `chunk_tx` and ends;
-    /// the handler future is awaited to completion (it is not yet
-    /// cancelled — the abort-cancels-Pub mechanism is review 001 Unit 9).
+    /// `PublishStream` while we feed it. The feed and the handler are
+    /// raced via `tokio::select!` (P-07): if the handler returns early
+    /// (e.g. rejects the upload after chunk 1), the feed is
+    /// short-circuited and the response is written immediately, without
+    /// waiting for the initiator to finish publishing. Dropping
+    /// `chunk_tx` on the handler-wins branch signals the feed to stop;
+    /// the handler's `PublishStream` sees EOF on its next `next().await`.
+    ///
+    /// Per-chunk validation (P-03, ADR-046 §4): when
+    /// `sink.publish_validator` is `Some`, each `call.published` chunk's
+    /// `input` is validated against the schema before being sent into
+    /// `chunk_tx`. On validation failure, an
+    /// `Err(CallError::invalid_input(...))` is injected and the feed
+    /// terminates — matching the `SinkHandler` contract that an `Err`
+    /// terminates the stream. When `publish_validator` is `None`,
+    /// chunks are yielded as-is (current behavior).
+    ///
+    /// Initiator `call.error` (P-04, ADR-046 §6): a `call.error` event
+    /// from the initiator is parsed into a `CallError`, injected as an
+    /// `Err` item, and terminates the feed — the handler sees the
+    /// initiator's error, not a synthetic "aborted" message.
+    ///
+    /// On `call.aborted` for this request ID, the feed injects an `Err`
+    /// into `chunk_tx` and ends; the handler future is awaited to
+    /// completion (it is not yet cancelled — the abort-cancels-Pub
+    /// mechanism is review 001 Unit 9).
     pub(crate) async fn pump_sink<R, W>(
         &self,
         reader: &mut super::wire::FrameFramedReader<R>,
@@ -458,58 +515,100 @@ impl Dispatcher {
         let SinkDispatch {
             handler,
             mut chunk_tx,
+            publish_validator,
         } = 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;
+        use futures::FutureExt;
+        let mut handler = handler.fuse();
+        let mut response: Option<ResponseEnvelope> = None;
+
+        loop {
+            tokio::select! {
+                biased;
+                resp = &mut handler => {
+                    response = Some(resp);
+                    break;
+                }
+                read = reader.read_frame() => {
+                    match read {
+                        Err(super::wire::FrameError::ConnectionClosed) => break,
+                        Err(err) => {
+                            warn!(error = %err, "frame read error during sink pump; ending stream");
+                            break;
                         }
-                        match envelope.r#type.as_str() {
-                            EVENT_PUBLISHED => {
-                                let chunk = envelope
-                                    .payload
-                                    .get("input")
-                                    .cloned()
-                                    .unwrap_or(Value::Null);
-                                if chunk_tx.send(Ok(chunk)).await.is_err() {
-                                    break;
-                                }
-                            }
-                            EVENT_COMPLETED => break,
-                            EVENT_ABORTED => {
-                                let _ = chunk_tx
-                                    .send(Err(CallError::internal("publish aborted by initiator")))
-                                    .await;
-                                break;
-                            }
-                            _ => {
+                        Ok(envelope) => {
+                            if envelope.id != request_id {
                                 debug!(
                                     event_type = %envelope.r#type,
-                                    "ignoring non-published event during sink pump"
+                                    id = %envelope.id,
+                                    "ignoring event with mismatched id during sink pump"
                                 );
+                                continue;
+                            }
+                            match envelope.r#type.as_str() {
+                                EVENT_PUBLISHED => {
+                                    let chunk = envelope
+                                        .payload
+                                        .get("input")
+                                        .cloned()
+                                        .unwrap_or(Value::Null);
+                                    if let Some(validator) = &publish_validator {
+                                        if !validator.is_valid(&chunk) {
+                                            let details = serde_json::json!({
+                                                "chunk": chunk,
+                                            });
+                                            let err = CallError::invalid_input(
+                                                "published chunk failed publish_schema validation",
+                                            )
+                                            .with_details(details);
+                                            let _ = chunk_tx.send(Err(err)).await;
+                                            break;
+                                        }
+                                    }
+                                    if chunk_tx.send(Ok(chunk)).await.is_err() {
+                                        break;
+                                    }
+                                }
+                                EVENT_COMPLETED => break,
+                                EVENT_ABORTED => {
+                                    let _ = chunk_tx
+                                        .send(Err(CallError::internal(
+                                            "publish aborted by initiator",
+                                        )))
+                                        .await;
+                                    break;
+                                }
+                                EVENT_ERROR => {
+                                    let call_error: CallError =
+                                        serde_json::from_value(envelope.payload).unwrap_or_else(
+                                            |_| {
+                                                CallError::internal(
+                                                    "publish error from initiator (malformed)",
+                                                )
+                                            },
+                                        );
+                                    let _ = chunk_tx.send(Err(call_error)).await;
+                                    break;
+                                }
+                                other => {
+                                    debug!(
+                                        event_type = %other,
+                                        "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;
+        drop(chunk_tx);
+
+        let response = match response {
+            Some(r) => r,
+            None => handler.await,
+        };
 
         let event: EventEnvelope = response.into();
         if let Err(err) = writer.write_frame(&event).await {
@@ -627,8 +726,7 @@ impl Dispatcher {
         });
 
         let mut reader = FrameFramedReader::new(reader);
-        let mut in_flight_sinks: HashMap<String, mpsc::Sender<Result<Value, CallError>>> =
-            HashMap::new();
+        let mut in_flight_sinks: HashMap<String, InFlightSink> = HashMap::new();
 
         loop {
             let envelope = match reader.read_frame().await {
@@ -664,8 +762,18 @@ impl Dispatcher {
                                 .await;
                         }
                         DispatchResult::Sink(sink) => {
-                            let SinkDispatch { handler, chunk_tx } = sink;
-                            in_flight_sinks.insert(request_id.clone(), chunk_tx);
+                            let SinkDispatch {
+                                handler,
+                                chunk_tx,
+                                publish_validator,
+                            } = sink;
+                            in_flight_sinks.insert(
+                                request_id.clone(),
+                                InFlightSink {
+                                    chunk_tx,
+                                    publish_validator,
+                                },
+                            );
                             let writer_clone = Arc::clone(&writer);
                             let request_id_for_handler = request_id.clone();
                             tokio::spawn(async move {
@@ -684,8 +792,9 @@ impl Dispatcher {
                 }
                 EVENT_ABORTED => {
                     let request_id = envelope.id.clone();
-                    if let Some(mut chunk_tx) = in_flight_sinks.remove(&request_id) {
-                        let _ = chunk_tx
+                    if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
+                        let _ = entry
+                            .chunk_tx
                             .send(Err(CallError::internal("publish aborted by initiator")))
                             .await;
                     } else {
@@ -699,9 +808,26 @@ impl Dispatcher {
                         .get("input")
                         .cloned()
                         .unwrap_or(Value::Null);
-                    if let Some(mut chunk_tx) = in_flight_sinks.remove(&request_id) {
-                        let _ = chunk_tx.send(Ok(chunk)).await;
-                        in_flight_sinks.insert(request_id, chunk_tx);
+                    if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
+                        let validated = match &entry.publish_validator {
+                            Some(validator) => {
+                                if validator.is_valid(&chunk) {
+                                    Ok(chunk)
+                                } else {
+                                    let details = serde_json::json!({ "chunk": chunk });
+                                    Err(CallError::invalid_input(
+                                        "published chunk failed publish_schema validation",
+                                    )
+                                    .with_details(details))
+                                }
+                            }
+                            None => Ok(chunk),
+                        };
+                        let keep = validated.is_ok();
+                        let _ = entry.chunk_tx.send(validated).await;
+                        if keep {
+                            in_flight_sinks.insert(request_id, entry);
+                        }
                     } else {
                         debug!(
                             request_id = %request_id,
@@ -713,6 +839,21 @@ impl Dispatcher {
                     let request_id = envelope.id.clone();
                     in_flight_sinks.remove(&request_id);
                 }
+                EVENT_ERROR => {
+                    let request_id = envelope.id.clone();
+                    if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
+                        let call_error: CallError = serde_json::from_value(envelope.payload)
+                            .unwrap_or_else(|_| {
+                                CallError::internal("publish error from initiator (malformed)")
+                            });
+                        let _ = entry.chunk_tx.send(Err(call_error)).await;
+                    } else {
+                        debug!(
+                            request_id = %request_id,
+                            "single-stream: call.error for unknown in-flight sink; dropping"
+                        );
+                    }
+                }
                 other => {
                     debug!(
                         event_type = %other,
@@ -1854,4 +1995,385 @@ mod tests {
             Some(&Value::String("NOT_FOUND".into()))
         );
     }
+
+    // --- ADR-046 §4: publish_schema validation (P-03) ---------------------
+
+    fn pub_spec_with_publish_schema(name: &str, publish_schema: Value) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Pub,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            AccessControl::default(),
+            None,
+        )
+        .with_publish_schema(publish_schema)
+    }
+
+    fn registry_with_pub_and_schema(
+        name: &str,
+        publish_schema: Value,
+        handler: crate::registry::registration::SinkHandler,
+    ) -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec_with_publish_schema(name, publish_schema),
+                HandlerKind::Sink(handler),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
+    /// A sink handler that records every item it received (`Ok` values and
+    /// the first `Err`) so tests can assert what the dispatch path injected
+    /// into the `PublishStream`. Returns the recordings in the response
+    /// output as `{ ok: [...], err: Option<CallError> }`.
+    fn recording_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 oks: Vec<Value> = Vec::new();
+            let mut err: Option<CallError> = None;
+            while let Some(item) = stream.next().await {
+                match item {
+                    Ok(v) => oks.push(v),
+                    Err(e) => {
+                        err = Some(e);
+                        break;
+                    }
+                }
+            }
+            ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "ok": oks, "err": err }))
+        })
+    }
+
+    #[tokio::test]
+    async fn handle_stream_pub_publish_schema_validates_and_rejects_invalid_chunk() {
+        let schema = serde_json::json!({
+            "type": "object",
+            "properties": { "bytes": { "type": "string" } },
+            "required": ["bytes"]
+        });
+        let registry = registry_with_pub_and_schema("fs/upload", schema, recording_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-schema-1",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": { "path": "/x" },
+            }),
+        );
+        let valid_chunk =
+            EventEnvelope::published("pub-schema-1", serde_json::json!({"bytes": "hello"}));
+        let invalid_chunk =
+            EventEnvelope::published("pub-schema-1", serde_json::json!({"bytes": 42}));
+        let completed = EventEnvelope::completed("pub-schema-1");
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&valid_chunk));
+        frame_buf.extend_from_slice(&encode_frame(&invalid_chunk));
+        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, "one response frame");
+        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
+        assert_eq!(frames[0].id, "pub-schema-1");
+        let output = frames[0].payload.get("output").expect("output field");
+        let oks = output
+            .get("ok")
+            .and_then(|v| v.as_array())
+            .expect("ok array");
+        assert_eq!(oks.len(), 1, "only the valid chunk reached the handler");
+        assert_eq!(oks[0], serde_json::json!({"bytes": "hello"}));
+        let err = output.get("err").expect("err field");
+        assert!(
+            err.get("code").and_then(|c| c.as_str()) == Some("INVALID_INPUT"),
+            "invalid chunk injects INVALID_INPUT, got: {err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn handle_stream_pub_publish_schema_passes_valid_chunks_unchanged() {
+        let schema = serde_json::json!({
+            "type": "object",
+            "properties": { "n": { "type": "integer" } },
+            "required": ["n"]
+        });
+        let registry = registry_with_pub_and_schema("fs/upload", schema, recording_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-schema-2",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": {},
+            }),
+        );
+        let c1 = EventEnvelope::published("pub-schema-2", serde_json::json!({"n": 1}));
+        let c2 = EventEnvelope::published("pub-schema-2", serde_json::json!({"n": 2}));
+        let c3 = EventEnvelope::published("pub-schema-2", serde_json::json!({"n": 3}));
+        let completed = EventEnvelope::completed("pub-schema-2");
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&c1));
+        frame_buf.extend_from_slice(&encode_frame(&c2));
+        frame_buf.extend_from_slice(&encode_frame(&c3));
+        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);
+        let output = frames[0].payload.get("output").expect("output field");
+        let oks = output
+            .get("ok")
+            .and_then(|v| v.as_array())
+            .expect("ok array");
+        assert_eq!(oks.len(), 3, "all three valid chunks reached the handler");
+        assert!(output
+            .get("err")
+            .and_then(|e| e.as_object())
+            .is_none_or(|o| o.is_empty()));
+    }
+
+    #[tokio::test]
+    async fn handle_stream_pub_no_publish_schema_yields_chunks_as_is() {
+        let registry = registry_with_pub("fs/upload", recording_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-schema-3",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": {},
+            }),
+        );
+        let c1 = EventEnvelope::published("pub-schema-3", serde_json::json!({"anything": true}));
+        let completed = EventEnvelope::completed("pub-schema-3");
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&c1));
+        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);
+        let output = frames[0].payload.get("output").expect("output field");
+        let oks = output
+            .get("ok")
+            .and_then(|v| v.as_array())
+            .expect("ok array");
+        assert_eq!(oks.len(), 1);
+        assert_eq!(oks[0], serde_json::json!({"anything": true}));
+    }
+
+    // --- ADR-046 §6: initiator call.error terminates the stream (P-04) ----
+
+    #[tokio::test]
+    async fn handle_stream_pub_initiator_call_error_terminates_stream_with_initiator_error() {
+        let registry = registry_with_pub("fs/upload", recording_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-err-1",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": {},
+            }),
+        );
+        let c1 = EventEnvelope::published("pub-err-1", serde_json::json!({"chunk": 1}));
+        let initiator_error = EventEnvelope::error(
+            "pub-err-1",
+            &CallError::new("UPLOAD_FAILED", "disk full on initiator", false),
+        );
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&c1));
+        frame_buf.extend_from_slice(&encode_frame(&initiator_error));
+
+        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, "one response frame");
+        let output = frames[0].payload.get("output").expect("output field");
+        let oks = output
+            .get("ok")
+            .and_then(|v| v.as_array())
+            .expect("ok array");
+        assert_eq!(oks.len(), 1, "the one valid chunk before the error");
+        let err = output.get("err").expect("err field");
+        assert_eq!(
+            err.get("code").and_then(|c| c.as_str()),
+            Some("UPLOAD_FAILED"),
+            "the initiator's CallError code reaches the handler, not a synthetic 'aborted'"
+        );
+        assert_eq!(
+            err.get("message").and_then(|m| m.as_str()),
+            Some("disk full on initiator")
+        );
+    }
+
+    #[tokio::test]
+    async fn handle_stream_pub_initiator_call_error_malformed_payload_falls_back_to_internal() {
+        let registry = registry_with_pub("fs/upload", recording_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-err-2",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": {},
+            }),
+        );
+        let malformed_error = EventEnvelope::new(
+            "call.error",
+            "pub-err-2",
+            serde_json::json!("not an object"),
+        );
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&malformed_error));
+
+        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);
+        let output = frames[0].payload.get("output").expect("output field");
+        let err = output.get("err").expect("err field");
+        assert_eq!(
+            err.get("code").and_then(|c| c.as_str()),
+            Some("INTERNAL"),
+            "malformed call.error payload falls back to INTERNAL"
+        );
+    }
+
+    // --- ADR-046 §7: early handler return short-circuits the feed (P-07) --
+
+    #[tokio::test]
+    async fn handle_stream_pub_early_handler_return_short_circuits_slow_feed() {
+        use std::sync::Arc as StdArc;
+        use std::sync::Mutex as StdMutex;
+
+        let seen_chunks: StdArc<StdMutex<Vec<Value>>> = StdArc::new(StdMutex::new(Vec::new()));
+        let seen_clone = StdArc::clone(&seen_chunks);
+        let early_handler: crate::registry::registration::SinkHandler = {
+            use crate::registry::registration::make_sink_handler;
+            use futures::stream::StreamExt;
+            make_sink_handler(move |_input, ctx, mut stream| {
+                let seen = StdArc::clone(&seen_clone);
+                async move {
+                    if let Some(Ok(v)) = stream.next().await {
+                        seen.lock().unwrap().push(v);
+                    }
+                    ResponseEnvelope::ok(
+                        ctx.request_id,
+                        serde_json::json!({ "rejected": true, "after": 1 }),
+                    )
+                }
+            })
+        };
+        let registry = registry_with_pub("fs/upload", early_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-early-1",
+            serde_json::json!({
+                "operationId": "/fs/upload",
+                "input": {},
+            }),
+        );
+        let c1 = EventEnvelope::published("pub-early-1", serde_json::json!({"chunk": 1}));
+
+        let mut frame_buf = Vec::new();
+        frame_buf.extend_from_slice(&encode_frame(&request));
+        frame_buf.extend_from_slice(&encode_frame(&c1));
+
+        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 start = std::time::Instant::now();
+        dp.handle_stream(conn, stream).await;
+        let elapsed = start.elapsed();
+
+        let frames = read_all_frames(&mut sink).await;
+        assert_eq!(frames.len(), 1, "response written once handler returns");
+        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
+        assert_eq!(frames[0].id, "pub-early-1");
+        let output = frames[0].payload.get("output").expect("output field");
+        assert_eq!(output["rejected"], serde_json::json!(true));
+        assert!(
+            elapsed < std::time::Duration::from_secs(5),
+            "response must not be deferred behind a slow/absent feed (elapsed: {elapsed:?})"
+        );
+        let seen = seen_chunks.lock().unwrap();
+        assert_eq!(
+            seen.len(),
+            1,
+            "handler consumed exactly one chunk before returning"
+        );
+    }
 }