//! Shared dispatch loop for `alk/call` connections.
//!
//! Both [`super::adapter::CallAdapter`]'s accept path and
//! [`crate::client::CallClient`]'s connect path produce a [`CallConnection`]
//! and hand it to the same dispatch loop here (ADR-017 §1): the loop reads
//! `EventEnvelope` frames off accepted bidirectional streams, dispatches
//! `call.requested` events against the operation registry, and writes the
//! response back on the same stream. The connection-establishment half
//! differs (accept vs dial); the dispatch half is shared.
//!
//! 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 parking_lot::Mutex as ParkingLotMutex;
use serde_json::Value;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tracing::{debug, warn};

use super::abort::AbortCascade;
use super::connection::CallConnection;
use super::wire::{
    CallError, EventEnvelope, FrameFramedReader, FrameFramedWriter, ResponseEnvelope,
    EVENT_ABORTED, EVENT_COMPLETED, EVENT_ERROR, EVENT_PUBLISHED, EVENT_REQUESTED, EVENT_RESPONDED,
};
use crate::protocol::adapter::SessionOverlaySource;
use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
use crate::registry::env::{LocalOperationEnv, OperationEnv, PeerCompositeEnv};
use crate::registry::registration::{OperationRegistry, PublishStream, ResponseStream};
use crate::registry::spec::OperationType;

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-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`), 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`.
/// - `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
/// (`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(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>,
    /// Marked (`channel_open`) `Pub` op (ADR-047 amendment 4 §4.1):
    /// publish chunks never feed the handler — the first
    /// `call.published` chunk is rejected loudly in the initiator's
    /// direction and the rest are ignored.
    pub(crate) reject_publish: bool,
}

/// 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`, the per-chunk validator (built once at dispatch
/// time from the op's `publish_schema`), and the spawned handler task's
/// `JoinHandle` so that `call.aborted` can cancel the handler future
/// (ADR-046 §5, review 001 Unit 9). `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>,
    handler_handle: JoinHandle<()>,
    /// Marked (`channel_open`) `Pub` op (ADR-047 amendment 4 §4.1):
    /// the first `call.published` chunk is rejected loudly in the
    /// initiator's direction and never fed to the handler.
    reject_publish: bool,
    reject_written: bool,
}

impl std::fmt::Debug for DispatchResult {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DispatchResult::Once(env) => f.debug_tuple("Once").field(env).finish(),
            DispatchResult::Stream(_) => {
                f.debug_tuple("Stream").field(&"<ResponseStream>").finish()
            }
            DispatchResult::Sink(_) => f.debug_tuple("Sink").field(&"<SinkDispatch>").finish(),
        }
    }
}

/// The started half of a dispatched `call.requested` event, from
/// `Dispatcher::dispatch_start`. `Once`/`Stream` are returned ready
/// for the caller to spawn (review 005 G-01); `Sink` was started
/// inline (its `chunk_tx` must be registered before the next
/// `call.published` frame is read) and carries the started handler
/// future.
pub enum StartedDispatch {
    Once(Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>),
    Stream(ResponseStream),
    Sink(SinkDispatch),
}

/// Shared dispatcher for an established `CallConnection`. Constructed by
/// both `CallAdapter` (accept path) and `CallClient` (connect path) and used
/// to run the dispatch loop. Holds no per-connection state; the
/// `CallConnection` is passed into `run_loop`.
///
/// `in_flight_sink_aborts` maps in-flight Pub `request_id`s to
/// `oneshot::Sender`s so that a cross-stream `call.aborted` (arriving on a
/// different `BiStream` than the one the Pub is pumping) can signal the
/// running `pump_sink` to cancel the handler future (ADR-046 §5, review 001
/// Unit 9). The map is shared across all `handle_stream` tasks spawned by
/// `run_loop`; each `pump_sink` inserts its abort sender on entry and
/// removes it on exit.
pub struct Dispatcher {
    pub registry: Arc<OperationRegistry>,
    pub identity_provider: Arc<dyn IdentityProvider>,
    pub session_source: Option<Arc<dyn SessionOverlaySource + Send + Sync>>,
    pub ownership_provider: Option<Arc<dyn OwnershipProvider>>,
    pub default_timeout: Duration,
    in_flight_sink_aborts: Arc<ParkingLotMutex<HashMap<String, oneshot::Sender<()>>>>,
}

impl Dispatcher {
    pub fn new(
        registry: Arc<OperationRegistry>,
        identity_provider: Arc<dyn IdentityProvider>,
    ) -> Self {
        Self {
            registry,
            identity_provider,
            session_source: None,
            ownership_provider: None,
            default_timeout: DEFAULT_TIMEOUT,
            in_flight_sink_aborts: Arc::new(ParkingLotMutex::new(HashMap::new())),
        }
    }

    pub fn with_session_source(
        mut self,
        source: Arc<dyn SessionOverlaySource + Send + Sync>,
    ) -> Self {
        self.session_source = Some(source);
        self
    }

    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.default_timeout = timeout;
        self
    }

    pub fn with_ownership_provider(mut self, provider: Arc<dyn OwnershipProvider>) -> Self {
        self.ownership_provider = Some(provider);
        self
    }

    fn strip_leading_slash(operation_id: &str) -> &str {
        operation_id.strip_prefix('/').unwrap_or(operation_id)
    }

    pub(crate) fn resolve_identity(
        &self,
        connection_identity: Option<Identity>,
        payload: &Value,
    ) -> Option<Identity> {
        let auth_token = payload.get("auth_token").and_then(|v| v.as_str());
        match auth_token {
            Some(token_str) => {
                let token = AuthToken {
                    raw: token_str.as_bytes().to_vec(),
                };
                match self.identity_provider.resolve_from_token(&token) {
                    Some(identity) => Some(identity),
                    None => connection_identity,
                }
            }
            None => connection_identity,
        }
    }

    pub fn compose_root_env(
        &self,
        connection: &CallConnection,
        context: &OperationContext,
    ) -> Arc<dyn OperationEnv + Send + Sync> {
        let base: Arc<dyn OperationEnv + Send + Sync> =
            Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));
        let session = self
            .session_source
            .as_ref()
            .and_then(|s| s.overlay_for(context));

        let mut env = PeerCompositeEnv::new(base);
        if let Some(session) = session {
            env = env.with_session(session);
        }
        if let Some(peer_id) = connection.identity().map(|identity| identity.id.clone()) {
            env.attach_peer(peer_id, connection.overlay_env());
        }
        Arc::new(env)
    }

    pub(crate) fn build_root_context(
        &self,
        request_id: String,
        operation_name: &str,
        identity: Option<Identity>,
        forwarded_for: Option<Identity>,
        connection: &CallConnection,
    ) -> OperationContext {
        let registration = self.registry.registration(operation_name);
        let (composition_authority, capabilities, scoped_env) = match registration {
            Some(r) => (
                r.composition_authority.clone(),
                r.capabilities.clone(),
                r.scoped_env.clone().unwrap_or_else(ScopedPeerEnv::empty),
            ),
            None => (
                None,
                crate::core::types::Capabilities::new(),
                ScopedPeerEnv::empty(),
            ),
        };

        let stub_env: Arc<dyn OperationEnv + Send + Sync> =
            Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));
        let mut context = OperationContext {
            request_id,
            parent_request_id: None,
            identity: identity.clone(),
            handler_identity: composition_authority,
            forwarded_for,
            capabilities,
            metadata: HashMap::new(),
            deadline: Some(Instant::now() + self.default_timeout),
            scoped_env,
            env: stub_env,
            abort_policy: AbortPolicy::default(),
            internal: false,
            ownership: self.ownership_provider.clone(),
        };
        context.env = self.compose_root_env(connection, &context);
        context
    }

    pub async fn dispatch_requested(
        &self,
        connection: &Arc<CallConnection>,
        request_id: String,
        payload: Value,
    ) -> ResponseEnvelope {
        let request_id_for_error = request_id.clone();
        match self.dispatch(connection, request_id, payload).await {
            DispatchResult::Once(envelope) => envelope,
            DispatchResult::Stream(mut stream) => stream.next().await.unwrap_or_else(|| {
                ResponseEnvelope::error(
                    request_id_for_error,
                    CallError::internal(
                        "dispatch_requested called on a Sub op; use the streaming path",
                    ),
                )
            }),
            DispatchResult::Sink(_) => ResponseEnvelope::error(
                request_id_for_error,
                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-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 `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>,
        request_id: String,
        payload: Value,
    ) -> DispatchResult {
        match self.dispatch_start(connection, request_id, payload) {
            StartedDispatch::Once(invoke) => DispatchResult::Once(invoke.await),
            StartedDispatch::Stream(stream) => DispatchResult::Stream(stream),
            StartedDispatch::Sink(sink) => DispatchResult::Sink(sink),
        }
    }

    /// Start dispatching a `call.requested` event without awaiting the
    /// invocation (review 005 G-01). The synchronous prefix of
    /// [`dispatch`](Self::dispatch) — identity resolution, root-context
    /// construction, op-type branch — is split from the invocation:
    ///
    /// - `Query`/`Mutation` return the invocation as a boxed future
    ///   (`StartedDispatch::Once`) for the caller to spawn. Awaiting it
    ///   inline is what deadlocked the serving loops: a handler that
    ///   nested-composes over the same connection blocks the one read
    ///   loop that must resolve the nested call's response frame.
    /// - `Sub` returns the [`ResponseStream`] synchronously
    ///   (`invoke_streaming` is a sync call) for the caller to pump in
    ///   a spawned task — an inline pump starves the same loop for the
    ///   stream's whole lifetime.
    /// - `Pub` runs the full sink start inline (`StartedDispatch::Sink`):
    ///   ACL failure resolves to a ready `Once` future (the response
    ///   envelope is already built), success carries the `chunk_tx` the
    ///   loop must register in `in_flight_sinks` *before* the next
    ///   `call.published` frame can be read — the insert cannot race
    ///   the feed, so the sink start stays synchronous by design.
    pub(crate) fn dispatch_start(
        &self,
        connection: &Arc<CallConnection>,
        request_id: String,
        payload: Value,
    ) -> StartedDispatch {
        let operation_id = payload
            .get("operationId")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let operation_name = Self::strip_leading_slash(operation_id).to_string();

        let connection_identity = connection.identity().cloned();
        let identity = self.resolve_identity(connection_identity, &payload);
        let forwarded_for = payload
            .get("forwarded_for")
            .and_then(|v| serde_json::from_value::<Identity>(v.clone()).ok());

        let mut input = payload.get("input").cloned().unwrap_or(Value::Null);

        let op_type = self
            .registry
            .registration(&operation_name)
            .map(|r| r.spec.op_type)
            .unwrap_or(OperationType::Query);

        let mut context = self.build_root_context(
            request_id,
            &operation_name,
            identity,
            forwarded_for,
            connection,
        );

        match op_type {
            OperationType::Query | OperationType::Mutation => {
                let registry = Arc::clone(&self.registry);
                let name = operation_name;
                StartedDispatch::Once(Box::pin(async move {
                    registry.invoke(&name, input, context).await
                }))
            }
            OperationType::Sub => {
                context.deadline = None;
                let stream = self
                    .registry
                    .invoke_streaming(&operation_name, input, context);
                StartedDispatch::Stream(stream)
            }
            OperationType::Pub => {
                context.deadline = None;
                match self
                    .registry
                    .resolve_sink_handler(&operation_name, &mut input, &mut context)
                {
                    Ok(sink_handler) => {
                        let reject_publish = self
                            .registry
                            .registration(&operation_name)
                            .is_some_and(|r| r.spec.channel_open.is_some());
                        let publish_validator = self.registry.publish_validator(&operation_name);
                        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);
                        StartedDispatch::Sink(SinkDispatch {
                            handler: Box::pin(handler),
                            chunk_tx,
                            publish_validator,
                            reject_publish,
                        })
                    }
                    Err(envelope) => StartedDispatch::Once(Box::pin(std::future::ready(envelope))),
                }
            }
        }
    }

    /// Spawn the Once invocation to write its single response frame on
    /// completion. The write failure is a warning, not a loop exit: the
    /// loop keeps reading (a dying transport surfaces on the next read
    /// as `ConnectionClosed`), matching the Sink arm's behavior.
    fn spawn_once_dispatch(
        writer: &Arc<super::connection::SharedFrameWriter>,
        request_id: String,
        invoke: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
    ) -> JoinHandle<()> {
        let writer = Arc::clone(writer);
        tokio::spawn(async move {
            let response = invoke.await;
            let event: EventEnvelope = response.into();
            if let Err(err) = writer.write_frame(&event).await {
                warn!(
                    error = %err,
                    request_id = %request_id,
                    "serving loop: failed to write Once response frame"
                );
            }
        })
    }

    /// Spawn a subscription's [`ResponseStream`] pump: each
    /// [`ResponseEnvelope`] becomes a `call.responded` / `call.error`
    /// frame; on natural end, a `call.completed` frame. The shared
    /// writer serializes frames so the pump's frames do not interleave
    /// with concurrent calls' frames.
    fn spawn_stream_pump(
        writer: &Arc<super::connection::SharedFrameWriter>,
        request_id: String,
        mut stream: ResponseStream,
    ) -> JoinHandle<()> {
        let writer = Arc::clone(writer);
        tokio::spawn(async move {
            let mut last_was_error = false;
            while let Some(envelope) = stream.next().await {
                last_was_error = envelope.result.is_err();
                let event: EventEnvelope = envelope.into();
                if let Err(err) = writer.write_frame(&event).await {
                    warn!(
                        error = %err,
                        request_id = %request_id,
                        "serving loop: failed to write streaming frame"
                    );
                    return;
                }
            }
            if !last_was_error {
                let completed = EventEnvelope::completed(&request_id);
                if let Err(err) = writer.write_frame(&completed).await {
                    warn!(
                        error = %err,
                        request_id = %request_id,
                        "serving loop: failed to write call.completed"
                    );
                }
            }
        })
    }

    /// Spawn a Pub's handler task: awaits the handler future and writes
    /// the single `call.responded` / `call.error` frame.
    fn spawn_sink_response_writer(
        writer: &Arc<super::connection::SharedFrameWriter>,
        request_id: String,
        handler: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
    ) -> JoinHandle<()> {
        let writer = Arc::clone(writer);
        tokio::spawn(async move {
            let response = handler.await;
            let event: EventEnvelope = response.into();
            if let Err(err) = writer.write_frame(&event).await {
                warn!(
                    error = %err,
                    request_id = %request_id,
                    "serving loop: failed to write sink response frame"
                );
            }
        })
    }

    pub async fn handle_abort(&self, connection: &Arc<CallConnection>, request_id: &str) {
        if let Some(tx) = self.in_flight_sink_aborts.lock().remove(request_id) {
            let _ = tx.send(());
            debug!(request_id, "signalled in-flight sink abort");
        }
        let mut pending = connection.pending().lock();
        let mut cascade = AbortCascade::new(&mut pending);
        let aborted = cascade.cascade_abort(request_id, AbortPolicy::AbortDependents);
        pending.handle_aborted(request_id);
        if !aborted.is_empty() {
            debug!(count = aborted.len(), "abort cascade evicted descendants");
        }
    }

    pub(crate) async fn handle_stream(
        &self,
        connection: Arc<CallConnection>,
        stream: crate::core::types::BiStream,
    ) {
        // `stream` is a `BiStream` (ADR-009) — `AsyncRead + AsyncWrite + Send
        // + Unpin`. Split into the read and write halves the call protocol's
        // frame reader/writer consume. The split is the stdlib idiom; no
        // per-handler wrapper.
        let (recv, send) = tokio::io::split(stream);
        let mut reader = FrameFramedReader::new(recv);
        let mut writer = FrameFramedWriter::new(send);

        loop {
            let envelope = match reader.read_frame().await {
                Ok(env) => env,
                Err(super::wire::FrameError::ConnectionClosed) => break,
                Err(err) => {
                    warn!(error = %err, "stream frame read error; closing stream");
                    break;
                }
            };

            match envelope.r#type.as_str() {
                EVENT_REQUESTED => {
                    let request_id = envelope.id.clone();
                    let payload = envelope.payload.clone();

                    match self
                        .dispatch(&connection, request_id.clone(), payload)
                        .await
                    {
                        DispatchResult::Once(response) => {
                            let event: EventEnvelope = response.into();
                            if let Err(err) = writer.write_frame(&event).await {
                                warn!(error = %err, "failed to write response frame; closing stream");
                                break;
                            }
                        }
                        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 => {
                    let request_id = envelope.id.clone();
                    self.handle_abort(&connection, &request_id).await;
                }
                other => {
                    debug!(event_type = %other, id = %envelope.id, "ignoring non-requested/non-aborted event on inbound stream");
                }
            }
        }
    }

    /// Pump a subscription's [`ResponseStream`] to the wire: each
    /// [`ResponseEnvelope`] becomes an [`EventEnvelope`] frame (`call.responded`
    /// for `Ok`, `call.error` for `Err`). On natural stream end (the stream
    /// returned `None` without the last item being an `Err`), write a
    /// `call.completed` frame. An `Err` envelope is terminal — the stream
    /// ends after it and we do NOT write `call.completed` (ADR-021 §6).
    ///
    /// If a frame write fails the pump stops early; the stream is dropped on
    /// return, releasing the handler's resources via `Drop` (ADR-016). A
    /// `call.aborted` for this request ID mutates the `PendingRequestMap`
    /// (via `handle_abort` on another stream) but does not yet cancel this
    /// task — the pump ends when the stream ends naturally or the I/O fails
    /// (the abort-cancels mechanism for Sub/Pub is review 001 Unit 9).
    pub(crate) async fn pump_stream<W: tokio::io::AsyncWrite + Unpin>(
        &self,
        writer: &mut super::wire::FrameFramedWriter<W>,
        request_id: &str,
        mut stream: ResponseStream,
    ) {
        let mut last_was_error = false;
        while let Some(envelope) = stream.next().await {
            last_was_error = envelope.result.is_err();
            let event: EventEnvelope = envelope.into();
            if let Err(err) = writer.write_frame(&event).await {
                warn!(error = %err, "failed to write streaming frame; closing stream");
                return;
            }
        }

        if !last_was_error {
            let completed = EventEnvelope::completed(request_id);
            if let Err(err) = writer.write_frame(&completed).await {
                warn!(error = %err, "failed to write call.completed");
            }
        }
    }

    /// 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. 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 (same-stream or
    /// cross-stream), the handler future is dropped — the abort
    /// cancels the Pub (ADR-046 §5, review 001 Unit 9). A cross-stream
    /// abort arrives via `handle_abort` on a different `BiStream` and
    /// signals through the shared `in_flight_sink_aborts` map; a
    /// same-stream abort arrives as a `call.aborted` frame on this
    /// stream. In both cases the handler future is dropped (not
    /// awaited), `chunk_tx` is dropped (the `PublishStream` sees EOF),
    /// and no response is written to the wire — the aborting initiator
    /// already knows the request is cancelled.
    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,
            publish_validator,
            reject_publish,
        } = sink;

        let (abort_tx, mut abort_rx) = oneshot::channel::<()>();
        self.in_flight_sink_aborts
            .lock()
            .insert(request_id.clone(), abort_tx);

        use futures::FutureExt;
        let mut handler = handler.fuse();
        let mut response: Option<ResponseEnvelope> = None;
        let mut aborted = false;
        let mut reject_written = false;

        loop {
            tokio::select! {
                biased;
                resp = &mut handler => {
                    response = Some(resp);
                    break;
                }
                _ = &mut abort_rx => {
                    aborted = true;
                    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;
                        }
                        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() {
                                EVENT_PUBLISHED => {
                                    // Marked ops: the data plane is the adopted
                                    // channel's `BiStream` — publish chunks are
                                    // rejected loudly, never consumed (ADR-047
                                    // amendment 4 §4.1). The handler continues on
                                    // the channel plane; the initiator's pending
                                    // call resolves with this rejection.
                                    if reject_publish {
                                        if !reject_written {
                                            reject_written = true;
                                            let rejection: EventEnvelope = ResponseEnvelope::error(
                                                request_id.clone(),
                                                CallError::new(
                                                    crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED,
                                                    "the data plane of a marked pub op is the adopted \
                                                     channel's BiStream — call.published chunks are \
                                                     rejected (ADR-047 amendment 4 §4.1)",
                                                    false,
                                                ),
                                            ).into();
                                            if let Err(err) = writer.write_frame(&rejection).await {
                                                warn!(error = %err, request_id = %request_id, "failed to write marked-op publish rejection");
                                            }
                                        }
                                        continue;
                                    }
                                    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 => {
                                    aborted = true;
                                    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"
                                    );
                                }
                            }
                        }
                    }
                }
            }
        }

        self.in_flight_sink_aborts.lock().remove(&request_id);
        drop(chunk_tx);

        if aborted {
            drop(handler);
            return;
        }

        let response = match response {
            Some(r) => r,
            None => handler.await,
        };

        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
    /// outstanding pending requests on close. Returns when the connection is
    /// closed (accept loop yields `ConnectionClosed`/`StreamClosed`/`Timeout`).
    pub async fn run_loop(self, connection: Arc<CallConnection>) {
        let pending = Arc::clone(connection.pending());

        let quic = match connection.connection() {
            Some(c) => Arc::clone(c),
            None => {
                warn!("run_loop called with an overlay-only CallConnection; returning");
                return;
            }
        };

        let sweeper_pending = Arc::clone(&pending);
        let sweeper_handle: JoinHandle<()> = tokio::spawn(async move {
            let mut interval = tokio::time::interval(SWEEPER_INTERVAL);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                interval.tick().await;
                let evicted = sweeper_pending.lock().evict_expired();
                if !evicted.is_empty() {
                    debug!(
                        count = evicted.len(),
                        "sweeper evicted expired pending entries"
                    );
                }
            }
        });

        loop {
            match quic.accept_bi().await {
                Ok(stream) => {
                    let conn = Arc::clone(&connection);
                    let dispatcher = self.clone();
                    tokio::spawn(async move {
                        dispatcher.handle_stream(conn, stream).await;
                    });
                }
                Err(StreamError::ConnectionClosed) => break,
                Err(StreamError::StreamClosed) => break,
                Err(StreamError::Timeout) => break,
                Err(err) => {
                    warn!(error = %err, "accept_bi error; stopping accept loop");
                    break;
                }
            }
        }

        let failed = pending
            .lock()
            .fail_all(CallError::internal("connection closed"));
        if !failed.is_empty() {
            debug!(
                count = failed.len(),
                "failed pending requests on connection close"
            );
        }

        sweeper_handle.abort();
    }

    /// Run the single-stream dispatch loop (ADR-036 amendment — channel
    /// 0 single-stream mode). Reads `EventEnvelope` frames off `reader`
    /// (channel 0's read half) and dispatches them, writing responses
    /// through `writer` (the shared frame writer). All frames —
    /// requests, responses, published chunks, aborts — are multiplexed
    /// on the one `BiStream` (channel 0).
    ///
    /// Unlike `run_loop` (stream-per-request), there is one read loop
    /// for the whole connection. In-flight Pub (Sink) operations are
    /// tracked by `request_id` in `in_flight_sinks`: their
    /// `call.published` / `call.completed` / `call.aborted` frames
    /// arrive *after* the `call.requested` and are routed to the
    /// matching sink's `chunk_tx` while new `call.requested` frames for
    /// other requests continue to be dispatched. Once invocations and
    /// Sub pumps are spawned (review 005 G-01 — inline dispatch
    /// deadlocked same-connection nested composition); only the sink
    /// start runs inline.
    ///
    /// This is the accept side's serving loop, so the loop also carries
    /// the pending-resolution arms the connect side's
    /// `serve_single_stream` composes: a nested-composing handler
    /// (e.g. a `from_call` imported-op stub riding this connection)
    /// blocks its own task, not the read loop, and the loop resolves
    /// the stub's response frames here. Without these arms the accept
    /// side could dispatch inbound requests but never resolve its own
    /// outbound pendings in single-stream mode.
    ///
    /// Returns when the read half closes (transport EOF). Outstanding
    /// pending requests are failed with `connection closed`, in-flight
    /// sinks' `chunk_tx` are dropped (the handler's `PublishStream`
    /// sees EOF), and spawned Once/Stream/Sink tasks are aborted.
    pub async fn run_loop_single_stream(
        self,
        connection: Arc<CallConnection>,
        reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
        writer: Arc<super::connection::SharedFrameWriter>,
    ) {
        let pending = Arc::clone(connection.pending());

        let sweeper_pending = Arc::clone(&pending);
        let sweeper_handle: JoinHandle<()> = tokio::spawn(async move {
            let mut interval = tokio::time::interval(SWEEPER_INTERVAL);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                interval.tick().await;
                let evicted = sweeper_pending.lock().evict_expired();
                if !evicted.is_empty() {
                    debug!(
                        count = evicted.len(),
                        "sweeper evicted expired pending entries"
                    );
                }
            }
        });

        let mut reader = FrameFramedReader::new(reader);
        let in_flight_sinks: Arc<ParkingLotMutex<HashMap<String, InFlightSink>>> =
            Arc::new(ParkingLotMutex::new(HashMap::new()));
        let spawned: Arc<ParkingLotMutex<Vec<JoinHandle<()>>>> =
            Arc::new(ParkingLotMutex::new(Vec::new()));

        loop {
            let envelope = match reader.read_frame().await {
                Ok(env) => env,
                Err(super::wire::FrameError::ConnectionClosed) => break,
                Err(err) => {
                    warn!(error = %err, "single-stream frame read error; closing loop");
                    break;
                }
            };

            match envelope.r#type.as_str() {
                EVENT_REQUESTED => {
                    let request_id = envelope.id.clone();
                    match self.dispatch_start(&connection, request_id.clone(), envelope.payload) {
                        StartedDispatch::Once(invoke) => {
                            let handle =
                                Self::spawn_once_dispatch(&writer, request_id.clone(), invoke);
                            spawned.lock().push(handle);
                        }
                        StartedDispatch::Stream(stream) => {
                            let handle =
                                Self::spawn_stream_pump(&writer, request_id.clone(), stream);
                            spawned.lock().push(handle);
                        }
                        StartedDispatch::Sink(sink) => {
                            let SinkDispatch {
                                handler,
                                chunk_tx,
                                publish_validator,
                                reject_publish,
                            } = sink;
                            let handle = Self::spawn_sink_response_writer(
                                &writer,
                                request_id.clone(),
                                handler,
                            );
                            in_flight_sinks.lock().insert(
                                request_id.clone(),
                                InFlightSink {
                                    chunk_tx,
                                    publish_validator,
                                    handler_handle: handle,
                                    reject_publish,
                                    reject_written: false,
                                },
                            );
                        }
                    }
                }
                EVENT_RESPONDED => {
                    let request_id = envelope.id.clone();
                    let output = envelope
                        .payload
                        .get("output")
                        .cloned()
                        .unwrap_or(Value::Null);
                    pending.lock().handle_responded(&request_id, output);
                }
                EVENT_COMPLETED => {
                    let request_id = envelope.id.clone();
                    if in_flight_sinks.lock().remove(&request_id).is_none() {
                        pending.lock().handle_completed(&request_id);
                    }
                }
                EVENT_ABORTED => {
                    let request_id = envelope.id.clone();
                    let entry = in_flight_sinks.lock().remove(&request_id);
                    if let Some(mut entry) = entry {
                        entry.handler_handle.abort();
                        let _ = entry
                            .chunk_tx
                            .send(Err(CallError::internal("publish aborted by initiator")))
                            .await;
                    } else {
                        self.handle_abort(&connection, &request_id).await;
                    }
                }
                EVENT_PUBLISHED => {
                    let request_id = envelope.id.clone();
                    let chunk = envelope
                        .payload
                        .get("input")
                        .cloned()
                        .unwrap_or(Value::Null);
                    let entry = in_flight_sinks.lock().remove(&request_id);
                    if let Some(mut entry) = entry {
                        if entry.reject_publish {
                            // Marked ops: the data plane is the adopted
                            // channel's `BiStream` — publish chunks are
                            // rejected loudly, never consumed (ADR-047
                            // amendment 4 §4.1). The loud error is written
                            // once per request; the entry stays registered
                            // so the handler's own final response still gets
                            // written. The handler continues on the channel
                            // plane; the initiator's pending call resolves
                            // with this rejection immediately.
                            if !entry.reject_written {
                                entry.reject_written = true;
                                let rejection: EventEnvelope = ResponseEnvelope::error(
                                    request_id.clone(),
                                    CallError::new(
                                        crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED,
                                        "the data plane of a marked pub op is the \
                                         adopted channel's BiStream — call.published \
                                         chunks are rejected (ADR-047 amendment 4 §4.1)",
                                        false,
                                    ),
                                )
                                .into();
                                if let Err(err) = writer.write_frame(&rejection).await {
                                    warn!(error = %err, request_id = %request_id, "failed to write marked-op publish rejection");
                                }
                            }
                            in_flight_sinks.lock().insert(request_id, entry);
                            continue;
                        }
                        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.lock().insert(request_id, entry);
                        }
                    } else {
                        debug!(
                            request_id = %request_id,
                            "single-stream: call.published for unknown in-flight sink; dropping"
                        );
                    }
                }
                EVENT_ERROR => {
                    let request_id = envelope.id.clone();
                    let call_error: CallError = serde_json::from_value(envelope.payload)
                        .unwrap_or_else(|_| {
                            CallError::internal("publish error from initiator (malformed)")
                        });
                    let entry = in_flight_sinks.lock().remove(&request_id);
                    if let Some(mut entry) = entry {
                        let _ = entry.chunk_tx.send(Err(call_error)).await;
                    } else {
                        pending.lock().handle_error(&request_id, call_error);
                    }
                }
                other => {
                    debug!(
                        event_type = %other,
                        id = %envelope.id,
                        "single-stream: ignoring unknown event type"
                    );
                }
            }
        }

        in_flight_sinks.lock().clear();
        for handle in spawned.lock().drain(..) {
            handle.abort();
        }

        let failed = pending
            .lock()
            .fail_all(CallError::internal("connection closed"));
        if !failed.is_empty() {
            debug!(
                count = failed.len(),
                "single-stream: failed pending requests on connection close"
            );
        }

        sweeper_handle.abort();
    }

    /// The full-duplex single-stream serving loop (review 004 F-04):
    /// composes the dispatch arms (`run_loop_single_stream`) and the
    /// pending-resolution arms (`read_single_stream_until_closed`) in
    /// one loop. Both sides of a channels connection can be both
    /// producer and consumer (AGENTS.md §8; ADR-022 §2) — on channel 0
    /// the two directions' frames are multiplexed on one byte stream,
    /// so the loop branches per frame:
    ///
    /// - `call.requested` → start the dispatch and spawn the work (the
    ///   serving half — what the accept side's `run_loop_single_stream`
    ///   does). Once invocations and Sub pumps run as spawned tasks;
    ///   only the Pub sink start runs inline (its `chunk_tx` must be
    ///   registered before the next `call.published` frame is read).
    ///   Inline dispatch was the review 005 G-01 defect: a handler that
    ///   nested-composes over the same connection blocked the one read
    ///   loop that must resolve the nested call's response frame. The
    ///   `SharedFrameWriter` serializes frames, so spawned arms' frames
    ///   do not interleave mid-frame.
    /// - `call.responded` / `call.completed` / `call.error` → resolve
    ///   the matching outbound pending entry (the calling half — what
    ///   the client read pump does). `call.responded` frames for
    ///   *inbound* Sub responses are `call.responded` too; direction is
    ///   disambiguated by the pending map (an id that is one of *our*
    ///   outbound pendings resolves there; an unknown id is dropped
    ///   with a debug line, never an error — the peer may legitimately
    ///   stream a Sub's responses that this side does not track).
    /// - `call.aborted` → try both tables: the in-flight sink aborts
    ///   *and* the outbound pending map's abort-cascade
    ///   (`handle_abort`), then fall through to serving-side sink
    ///   cancellation (the `run_loop_single_stream` arm) when the id
    ///   matches an inbound sink. An id in neither table is a no-op.
    /// - `call.published` / `call.completed` (initiator→responder) →
    ///   route to the matching inbound in-flight sink; ids that are
    ///   outbound pending *call* entries whose `call.responded` already
    ///   removed them are no-ops.
    ///
    /// IDs are UUID-generated per side (`generate_request_id()`), so
    /// cross-correlation between the two directions is not a hazard.
    ///
    /// On read-half close, outbound pendings are failed (`connection
    /// closed`), in-flight inbound sinks are dropped, and the spawned
    /// Once/Stream/Sink tasks are aborted (their response frames cannot
    /// reach a closed transport anyway).
    pub async fn serve_single_stream(
        self,
        connection: Arc<CallConnection>,
        reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
        writer: Arc<super::connection::SharedFrameWriter>,
    ) {
        let pending = Arc::clone(connection.pending());

        let sweeper_pending = Arc::clone(&pending);
        let sweeper_handle: JoinHandle<()> = tokio::spawn(async move {
            let mut interval = tokio::time::interval(SWEEPER_INTERVAL);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                interval.tick().await;
                let evicted = sweeper_pending.lock().evict_expired();
                if !evicted.is_empty() {
                    debug!(
                        count = evicted.len(),
                        "serve loop: sweeper evicted expired pending entries"
                    );
                }
            }
        });

        let mut reader = FrameFramedReader::new(reader);
        let in_flight_sinks: Arc<ParkingLotMutex<HashMap<String, InFlightSink>>> =
            Arc::new(ParkingLotMutex::new(HashMap::new()));
        let spawned: Arc<ParkingLotMutex<Vec<JoinHandle<()>>>> =
            Arc::new(ParkingLotMutex::new(Vec::new()));

        loop {
            let envelope = match reader.read_frame().await {
                Ok(env) => env,
                Err(super::wire::FrameError::ConnectionClosed) => break,
                Err(err) => {
                    warn!(error = %err, "serve loop: frame read error; closing loop");
                    break;
                }
            };

            match envelope.r#type.as_str() {
                EVENT_REQUESTED => {
                    let request_id = envelope.id.clone();
                    match self.dispatch_start(&connection, request_id.clone(), envelope.payload) {
                        StartedDispatch::Once(invoke) => {
                            let handle =
                                Self::spawn_once_dispatch(&writer, request_id.clone(), invoke);
                            spawned.lock().push(handle);
                        }
                        StartedDispatch::Stream(stream) => {
                            let handle =
                                Self::spawn_stream_pump(&writer, request_id.clone(), stream);
                            spawned.lock().push(handle);
                        }
                        StartedDispatch::Sink(sink) => {
                            let SinkDispatch {
                                handler,
                                chunk_tx,
                                publish_validator,
                                reject_publish,
                            } = sink;
                            let handle = Self::spawn_sink_response_writer(
                                &writer,
                                request_id.clone(),
                                handler,
                            );
                            in_flight_sinks.lock().insert(
                                request_id.clone(),
                                InFlightSink {
                                    chunk_tx,
                                    publish_validator,
                                    handler_handle: handle,
                                    reject_publish,
                                    reject_written: false,
                                },
                            );
                        }
                    }
                }
                EVENT_RESPONDED => {
                    let request_id = envelope.id.clone();
                    let output = envelope
                        .payload
                        .get("output")
                        .cloned()
                        .unwrap_or(Value::Null);
                    pending.lock().handle_responded(&request_id, output);
                }
                EVENT_COMPLETED => {
                    let request_id = envelope.id.clone();
                    if in_flight_sinks.lock().remove(&request_id).is_none() {
                        pending.lock().handle_completed(&request_id);
                    }
                }
                EVENT_ABORTED => {
                    let request_id = envelope.id.clone();
                    let entry = in_flight_sinks.lock().remove(&request_id);
                    if let Some(mut entry) = entry {
                        entry.handler_handle.abort();
                        let _ = entry
                            .chunk_tx
                            .send(Err(CallError::internal("publish aborted by initiator")))
                            .await;
                    } else {
                        self.handle_abort(&connection, &request_id).await;
                    }
                }
                EVENT_PUBLISHED => {
                    let request_id = envelope.id.clone();
                    let chunk = envelope
                        .payload
                        .get("input")
                        .cloned()
                        .unwrap_or(Value::Null);
                    let entry = in_flight_sinks.lock().remove(&request_id);
                    if let Some(mut entry) = entry {
                        if entry.reject_publish {
                            // Marked ops: the data plane is the adopted
                            // channel's `BiStream` — publish chunks are
                            // rejected loudly, never consumed (ADR-047
                            // amendment 4 §4.1). The loud error is written
                            // once per request; the entry stays registered
                            // so the handler's own final response still gets
                            // written. The handler continues on the channel
                            // plane; the initiator's pending call resolves
                            // with this rejection immediately.
                            if !entry.reject_written {
                                entry.reject_written = true;
                                let rejection: EventEnvelope = ResponseEnvelope::error(
                                    request_id.clone(),
                                    CallError::new(
                                        crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED,
                                        "the data plane of a marked pub op is the \
                                         adopted channel's BiStream — call.published \
                                         chunks are rejected (ADR-047 amendment 4 §4.1)",
                                        false,
                                    ),
                                )
                                .into();
                                if let Err(err) = writer.write_frame(&rejection).await {
                                    warn!(error = %err, request_id = %request_id, "failed to write marked-op publish rejection");
                                }
                            }
                            in_flight_sinks.lock().insert(request_id, entry);
                            continue;
                        }
                        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.lock().insert(request_id, entry);
                        }
                    } else {
                        debug!(
                            request_id = %request_id,
                            "serve loop: call.published for unknown in-flight sink; dropping"
                        );
                    }
                }
                EVENT_ERROR => {
                    let request_id = envelope.id.clone();
                    let call_error: CallError = serde_json::from_value(envelope.payload)
                        .unwrap_or_else(|_| {
                            CallError::internal("publish error from initiator (malformed)")
                        });
                    let entry = in_flight_sinks.lock().remove(&request_id);
                    if let Some(mut entry) = entry {
                        let _ = entry.chunk_tx.send(Err(call_error)).await;
                    } else {
                        pending.lock().handle_error(&request_id, call_error);
                    }
                }
                other => {
                    debug!(
                        event_type = %other,
                        id = %envelope.id,
                        "serve loop: ignoring unknown event type"
                    );
                }
            }
        }

        in_flight_sinks.lock().clear();
        for handle in spawned.lock().drain(..) {
            handle.abort();
        }

        let failed = pending
            .lock()
            .fail_all(CallError::internal("connection closed"));
        if !failed.is_empty() {
            debug!(
                count = failed.len(),
                "serve loop: failed pending requests on connection close"
            );
        }

        sweeper_handle.abort();
    }
}

impl Clone for Dispatcher {
    fn clone(&self) -> Self {
        Self {
            registry: Arc::clone(&self.registry),
            identity_provider: Arc::clone(&self.identity_provider),
            session_source: self.session_source.clone(),
            ownership_provider: self.ownership_provider.clone(),
            default_timeout: self.default_timeout,
            in_flight_sink_aborts: Arc::clone(&self.in_flight_sink_aborts),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::auth::{AuthToken, Identity, IdentityProvider};
    use crate::core::types::Capabilities;
    use crate::protocol::wire::{EVENT_COMPLETED, EVENT_ERROR, EVENT_RESPONDED};
    use crate::registry::registration::{
        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
    };
    use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
    use std::collections::HashMap;
    use std::sync::Mutex as StdMutex;

    use crate::protocol::sink_empty_connection as stub_connection;

    struct StaticIdentityProvider {
        tokens: StdMutex<HashMap<String, Identity>>,
    }

    impl StaticIdentityProvider {
        fn new() -> Self {
            Self {
                tokens: StdMutex::new(HashMap::new()),
            }
        }

        fn with_token(self, token: &str, identity: Identity) -> Self {
            self.tokens
                .lock()
                .unwrap()
                .insert(token.to_string(), identity);
            self
        }
    }

    impl IdentityProvider for StaticIdentityProvider {
        fn resolve_from_fingerprint(&self, _fp: &str) -> Option<Identity> {
            None
        }
        fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
            let token_str = String::from_utf8_lossy(&token.raw);
            self.tokens.lock().unwrap().get(token_str.as_ref()).cloned()
        }
    }

    fn identity_with_scopes(id: &str, scopes: &[&str]) -> Identity {
        Identity {
            id: id.to_string(),
            scopes: scopes.iter().map(|s| s.to_string()).collect(),
            resources: HashMap::new(),
        }
    }

    fn external_spec(name: &str, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn registry_with(name: &str, visibility: Visibility, acl: AccessControl) -> OperationRegistry {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    name,
                    OperationType::Query,
                    visibility,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    acl,
                    None,
                ),
                HandlerKind::Once(make_handler(|input, context| async move {
                    ResponseEnvelope::ok(context.request_id, input)
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
    }

    fn dispatcher() -> Dispatcher {
        Dispatcher::new(
            Arc::new(OperationRegistry::new()),
            Arc::new(StaticIdentityProvider::new()),
        )
    }

    #[tokio::test]
    async fn dispatch_authorized_peer_dispatches_and_populates_capabilities() {
        let caps = Capabilities::new().with_api_key("google", "k".to_string());
        let registry = OperationRegistry::new();
        let handler = make_handler(|_input, context| async move {
            let has_google = context.capabilities.get("google").is_some();
            ResponseEnvelope::ok(
                context.request_id,
                serde_json::json!({ "has_google": has_google }),
            )
        });
        registry
            .register(HandlerRegistration::new(
                external_spec("admin/run", AccessControl::default()),
                HandlerKind::Once(handler),
                OperationProvenance::Local,
                None,
                None,
                caps,
            ))
            .unwrap();
        let registry = Arc::new(registry);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/admin/run",
            "input": {},
        });
        let response = dp
            .dispatch_requested(&conn, "req-1".to_string(), payload)
            .await;
        let out = response.result.expect("dispatch ok");
        assert_eq!(out["has_google"], Value::Bool(true));
    }

    #[tokio::test]
    async fn dispatch_unauthorized_peer_returns_forbidden_capabilities_never_populated() {
        let caps = Capabilities::new().with_api_key("google", "k".to_string());
        let registry = OperationRegistry::new();
        let handler = make_handler(|_input, context| async move {
            let has_google = context.capabilities.get("google").is_some();
            ResponseEnvelope::ok(
                context.request_id,
                serde_json::json!({ "has_google": has_google }),
            )
        });
        registry
            .register(HandlerRegistration::new(
                external_spec(
                    "admin/run",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(handler),
                OperationProvenance::Local,
                None,
                None,
                caps,
            ))
            .unwrap();
        let registry = Arc::new(registry);
        let provider: Arc<dyn IdentityProvider> = Arc::new(
            StaticIdentityProvider::new()
                .with_token("alk_user", identity_with_scopes("regular-user", &["user"])),
        );
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/admin/run",
            "input": {},
            "auth_token": "alk_user",
        });
        let response = dp
            .dispatch_requested(&conn, "req-2".to_string(), payload)
            .await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("admin"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_internal_op_from_wire_returns_not_found_before_acl() {
        let registry = Arc::new(registry_with(
            "secret/op",
            Visibility::Internal,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/secret/op",
            "input": {},
        });
        let response = dp
            .dispatch_requested(&conn, "req-3".to_string(), payload)
            .await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "NOT_FOUND");
                assert!(e.message.contains("secret/op"));
            }
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_connection_with_no_identity_produces_no_peer_id_in_env() {
        let registry = Arc::new(registry_with(
            "fs/readFile",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = CallConnection::new(stub_connection());

        let context = dp.build_root_context("req-4".to_string(), "fs/readFile", None, None, &conn);

        assert!(
            context.identity.is_none(),
            "no connection identity → context.identity is None"
        );
        assert!(
            context.env.peer_ids().is_empty(),
            "no peer overlay attached when connection has no identity"
        );
    }

    #[tokio::test]
    async fn dispatch_connection_with_identity_attaches_peer_overlay_keyed_by_identity_id() {
        let registry = Arc::new(registry_with(
            "fs/readFile",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = CallConnection::new(stub_connection());
        conn.connection()
            .expect("quic connection present")
            .set_identity(identity_with_scopes("worker-a", &[]))
            .expect("identity not yet set");

        let context = dp.build_root_context("req-5".to_string(), "fs/readFile", None, None, &conn);

        assert_eq!(
            context.env.peer_ids(),
            vec!["worker-a".to_string()],
            "PeerId for connection comes from connection.identity().id"
        );
    }

    #[tokio::test]
    async fn dispatch_extract_forwarded_for_from_payload_into_context() {
        let registry = OperationRegistry::new();
        let handler = make_handler(|_input, context| async move {
            let forwarded_id = context.forwarded_for.as_ref().map(|i| i.id.clone());
            ResponseEnvelope::ok(
                context.request_id,
                serde_json::json!({ "forwarded_for_id": forwarded_id }),
            )
        });
        registry
            .register(HandlerRegistration::new(
                external_spec("fs/readFile", AccessControl::default()),
                HandlerKind::Once(handler),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/fs/readFile",
            "input": {},
            "forwarded_for": {
                "id": "alice",
                "scopes": ["fs:read"],
                "resources": {}
            },
        });
        let response = dp
            .dispatch_requested(&conn, "req-6".to_string(), payload)
            .await;
        let out = response.result.expect("ok");
        assert_eq!(out["forwarded_for_id"], Value::String("alice".into()));
    }

    #[tokio::test]
    async fn dispatch_without_forwarded_for_field_is_none() {
        let registry = OperationRegistry::new();
        let handler = make_handler(|_input, context| async move {
            let present = context.forwarded_for.is_some();
            ResponseEnvelope::ok(
                context.request_id,
                serde_json::json!({ "present": present }),
            )
        });
        registry
            .register(HandlerRegistration::new(
                external_spec("fs/readFile", AccessControl::default()),
                HandlerKind::Once(handler),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/fs/readFile",
            "input": {},
        });
        let response = dp
            .dispatch_requested(&conn, "req-7".to_string(), payload)
            .await;
        let out = response.result.expect("ok");
        assert_eq!(out["present"], Value::Bool(false));
    }

    #[tokio::test]
    async fn dispatch_default_access_control_dispatches_to_any_peer() {
        let registry = Arc::new(registry_with(
            "echo/run",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/echo/run",
            "input": { "msg": "hi" },
        });
        let response = dp
            .dispatch_requested(&conn, "req-8".to_string(), payload)
            .await;
        assert_eq!(response.result, Ok(serde_json::json!({ "msg": "hi" })));
    }

    #[test]
    fn dispatcher_helper_compiles_with_full_signature() {
        let _dp = dispatcher();
    }

    // --- non-QUIC (overlay-only) dispatch path ----------------------------

    fn overlay_only_connection(identity: Identity) -> Arc<CallConnection> {
        Arc::new(CallConnection::new_overlay_only(identity))
    }

    #[tokio::test]
    async fn dispatch_requested_works_with_overlay_only_connection() {
        let registry = Arc::new(registry_with(
            "echo/run",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));

        let payload = serde_json::json!({
            "operationId": "/echo/run",
            "input": { "msg": "hello" },
        });
        let response = dp
            .dispatch_requested(&conn, "ws-req-1".to_string(), payload)
            .await;
        assert_eq!(response.request_id, "ws-req-1");
        assert_eq!(response.result, Ok(serde_json::json!({ "msg": "hello" })));
    }

    #[tokio::test]
    async fn dispatch_requested_overlay_only_attaches_peer_keyed_by_stored_identity() {
        let registry = OperationRegistry::new();
        let handler = make_handler(|_input, context| async move {
            let peer_ids = context.env.peer_ids();
            ResponseEnvelope::ok(
                context.request_id,
                serde_json::json!({ "peer_ids": peer_ids }),
            )
        });
        registry
            .register(HandlerRegistration::new(
                external_spec("fs/readFile", AccessControl::default()),
                HandlerKind::Once(handler),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));

        let payload = serde_json::json!({
            "operationId": "/fs/readFile",
            "input": {},
        });
        let response = dp
            .dispatch_requested(&conn, "ws-req-2".to_string(), payload)
            .await;
        let out = response.result.expect("ok");
        assert_eq!(out["peer_ids"], serde_json::json!(["ws-peer"]));
    }

    #[tokio::test]
    async fn dispatch_requested_overlay_only_unknown_op_returns_not_found() {
        let registry = Arc::new(OperationRegistry::new());
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));

        let payload = serde_json::json!({
            "operationId": "/no/such/op",
            "input": {},
        });
        let response = dp
            .dispatch_requested(&conn, "ws-req-3".to_string(), payload)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn handle_abort_works_with_overlay_only_connection() {
        let registry = Arc::new(registry_with(
            "echo/run",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));

        let parent_id = "ws-abort-root".to_string();
        let child_id = "ws-abort-child".to_string();
        {
            let mut pending = conn.pending().lock();
            pending.register_call(
                parent_id.clone(),
                Some(Instant::now() + Duration::from_secs(30)),
                None,
            );
            pending.register_call(
                child_id.clone(),
                Some(Instant::now() + Duration::from_secs(30)),
                Some(parent_id.clone()),
            );
        }
        assert!(conn.pending().lock().contains(&parent_id));
        assert!(conn.pending().lock().contains(&child_id));

        dp.handle_abort(&conn, &parent_id).await;

        assert!(
            !conn.pending().lock().contains(&parent_id),
            "parent entry removed after abort"
        );
        assert!(
            !conn.pending().lock().contains(&child_id),
            "child aborted by cascade"
        );
    }

    #[tokio::test]
    async fn handle_abort_unknown_request_id_is_noop_for_overlay_only() {
        let registry = Arc::new(OperationRegistry::new());
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));

        dp.handle_abort(&conn, "totally-unknown").await;
        assert!(conn.pending().lock().is_empty());
    }

    #[tokio::test]
    async fn overlay_only_full_dispatch_round_trip_returns_response_envelope() {
        let registry = Arc::new(registry_with(
            "echo/run",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));

        let payload = serde_json::json!({
            "operationId": "/echo/run",
            "input": { "v": 42 },
        });
        let request_id = "ws-roundtrip-1".to_string();
        let response = dp
            .dispatch_requested(&conn, request_id.clone(), payload)
            .await;
        assert!(response.result.is_ok());
        let envelope: EventEnvelope = response.into();
        assert_eq!(envelope.r#type, EVENT_RESPONDED);
        assert_eq!(envelope.id, "ws-roundtrip-1");
        assert_eq!(
            envelope.payload.get("output"),
            Some(&serde_json::json!({ "v": 42 }))
        );
    }

    // --- streaming dispatch branch (ADR-021 §6) ---------------------------

    fn subscription_spec(name: &str, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Sub,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn encode_frame(envelope: &EventEnvelope) -> Vec<u8> {
        let body = serde_json::to_vec(envelope).expect("serialize envelope");
        let mut buf = (body.len() as u32).to_be_bytes().to_vec();
        buf.extend_from_slice(&body);
        buf
    }

    async fn read_all_frames(
        reader: &mut (impl tokio::io::AsyncRead + Unpin),
    ) -> Vec<EventEnvelope> {
        let mut buf = Vec::new();
        use tokio::io::AsyncReadExt;
        let _ = reader.read_to_end(&mut buf).await;
        let mut frames = Vec::new();
        let mut cursor = std::io::Cursor::new(buf);
        loop {
            let mut len_buf = [0u8; 4];
            match tokio::io::AsyncReadExt::read_exact(&mut cursor, &mut len_buf).await {
                Ok(_) => {}
                Err(_) => break,
            }
            let len = u32::from_be_bytes(len_buf) as usize;
            let mut body = vec![0u8; len];
            if tokio::io::AsyncReadExt::read_exact(&mut cursor, &mut body)
                .await
                .is_err()
            {
                break;
            }
            let envelope: EventEnvelope =
                serde_json::from_slice(&body).expect("deserialize written frame");
            frames.push(envelope);
        }
        frames
    }

    fn registry_with_subscription(
        name: &str,
        handler: crate::registry::registration::StreamingHandler,
    ) -> Arc<OperationRegistry> {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec(name, AccessControl::default()),
                HandlerKind::Stream(handler),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(registry)
    }

    #[tokio::test]
    async fn dispatch_subscription_returns_stream_result() {
        let handler = make_streaming_handler(|input, ctx| {
            futures::stream::iter(vec![
                ResponseEnvelope::ok(ctx.request_id.clone(), input.clone()),
                ResponseEnvelope::ok(ctx.request_id.clone(), serde_json::json!({"done": true})),
            ])
        });
        let registry = registry_with_subscription("events/stream", handler);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/events/stream",
            "input": { "v": 1 },
        });
        match dp.dispatch(&conn, "sub-1".to_string(), payload).await {
            DispatchResult::Stream(mut stream) => {
                use futures::stream::StreamExt;
                let first = stream.next().await.expect("first envelope");
                assert_eq!(first.request_id, "sub-1");
                assert_eq!(first.result, Ok(serde_json::json!({ "v": 1 })));
                let second = stream.next().await.expect("second envelope");
                assert_eq!(second.result, Ok(serde_json::json!({ "done": true })));
                assert!(
                    stream.next().await.is_none(),
                    "stream ends after two values"
                );
            }
            other => panic!("expected Stream, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_subscription_clears_deadline_to_none() {
        let handler = make_streaming_handler(|_input, ctx| {
            let deadline = ctx.deadline;
            futures::stream::iter(vec![ResponseEnvelope::ok(
                ctx.request_id.clone(),
                serde_json::json!({ "deadline_is_none": deadline.is_none() }),
            )])
        });
        let registry = registry_with_subscription("events/stream", handler);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/events/stream",
            "input": {},
        });
        match dp.dispatch(&conn, "sub-dl".to_string(), payload).await {
            DispatchResult::Stream(mut stream) => {
                use futures::stream::StreamExt;
                let env = stream.next().await.expect("one envelope");
                let out = env.result.expect("ok");
                assert_eq!(out["deadline_is_none"], Value::Bool(true));
            }
            other => panic!("expected Stream, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_query_keeps_deadline_some() {
        let registry = OperationRegistry::new();
        let handler = make_handler(|_input, ctx| async move {
            let deadline_is_some = ctx.deadline.is_some();
            ResponseEnvelope::ok(
                ctx.request_id.clone(),
                serde_json::json!({ "deadline_is_some": deadline_is_some }),
            )
        });
        registry
            .register(HandlerRegistration::new(
                external_spec("echo/run", AccessControl::default()),
                HandlerKind::Once(handler),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let payload = serde_json::json!({
            "operationId": "/echo/run",
            "input": {},
        });
        match dp.dispatch(&conn, "q-1".to_string(), payload).await {
            DispatchResult::Once(env) => {
                let out = env.result.expect("ok");
                assert_eq!(out["deadline_is_some"], Value::Bool(true));
            }
            other => panic!("expected Once, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn handle_stream_subscription_pumps_each_frame_then_completed() {
        let handler = make_streaming_handler(|input, ctx| {
            let first = input.clone();
            let rid = ctx.request_id.clone();
            futures::stream::iter(vec![
                ResponseEnvelope::ok(rid.clone(), first),
                ResponseEnvelope::ok(rid.clone(), serde_json::json!({"n": 2})),
                ResponseEnvelope::ok(rid, serde_json::json!({"n": 3})),
            ])
        });
        let registry = registry_with_subscription("events/stream", handler);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let request = EventEnvelope::requested(
            "sub-pump-1",
            serde_json::json!({
                "operationId": "/events/stream",
                "input": { "n": 1 },
            }),
        );
        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
        let (send, mut sink) = tokio::io::duplex(8 * 1024);
        let 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(), 4, "3 responded + 1 completed");
        for (i, f) in frames[..3].iter().enumerate() {
            assert_eq!(f.r#type, EVENT_RESPONDED, "frame {i} is call.responded");
            assert_eq!(f.id, "sub-pump-1");
        }
        assert_eq!(frames[3].r#type, EVENT_COMPLETED);
        assert_eq!(frames[3].id, "sub-pump-1");
        assert_eq!(frames[3].payload, serde_json::json!({}));
    }

    #[tokio::test]
    async fn handle_stream_subscription_error_is_terminal_no_completed() {
        let handler = make_streaming_handler(|_input, ctx| {
            let rid = ctx.request_id.clone();
            futures::stream::iter(vec![
                ResponseEnvelope::ok(rid.clone(), serde_json::json!({"ok": true})),
                ResponseEnvelope::error(rid.clone(), CallError::internal("boom")),
            ])
        });
        let registry = registry_with_subscription("events/stream", handler);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let request = EventEnvelope::requested(
            "sub-err-1",
            serde_json::json!({
                "operationId": "/events/stream",
                "input": {},
            }),
        );
        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
        let (send, mut sink) = tokio::io::duplex(8 * 1024);
        let 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(), 2, "1 responded + 1 error, no completed");
        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
        assert_eq!(frames[1].r#type, EVENT_ERROR);
        assert_eq!(frames[1].id, "sub-err-1");
        assert_eq!(
            frames[1].payload.get("code"),
            Some(&Value::String("INTERNAL".into()))
        );
    }

    #[tokio::test]
    async fn handle_stream_query_dispatch_unchanged_one_frame_no_completed() {
        let registry = Arc::new(registry_with(
            "echo/run",
            Visibility::External,
            AccessControl::default(),
        ));
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let request = EventEnvelope::requested(
            "q-pump-1",
            serde_json::json!({
                "operationId": "/echo/run",
                "input": { "msg": "hi" },
            }),
        );
        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
        let (send, mut sink) = tokio::io::duplex(8 * 1024);
        let 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, "query: exactly one frame, no completed");
        assert_eq!(frames[0].r#type, EVENT_RESPONDED);
        assert_eq!(frames[0].id, "q-pump-1");
        assert_eq!(
            frames[0].payload.get("output"),
            Some(&serde_json::json!({ "msg": "hi" }))
        );
    }

    #[tokio::test]
    async fn handle_stream_subscription_unknown_op_yields_single_error_no_completed() {
        let registry = Arc::new(OperationRegistry::new());
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let request = EventEnvelope::requested(
            "sub-missing-1",
            serde_json::json!({
                "operationId": "/no/such/stream",
                "input": {},
            }),
        );
        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
        let (send, mut sink) = tokio::io::duplex(8 * 1024);
        let 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 op: single error, no completed");
        assert_eq!(frames[0].r#type, EVENT_ERROR);
        assert_eq!(frames[0].id, "sub-missing-1");
        assert_eq!(
            frames[0].payload.get("code"),
            Some(&Value::String("NOT_FOUND".into()))
        );
    }

    #[tokio::test]
    async fn handle_stream_aborted_for_streaming_request_drops_stream() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc as StdArc;

        let dropped = StdArc::new(AtomicBool::new(false));
        let dropped_clone = StdArc::clone(&dropped);
        let handler = make_streaming_handler(move |_input, ctx| {
            let rid = ctx.request_id.clone();
            let flag = StdArc::clone(&dropped_clone);
            struct DropGuard(StdArc<AtomicBool>);
            impl Drop for DropGuard {
                fn drop(&mut self) {
                    self.0.store(true, Ordering::SeqCst);
                }
            }
            let guard = DropGuard(StdArc::clone(&flag));
            futures::stream::poll_fn(move |_cx| {
                if flag.load(Ordering::SeqCst) {
                    return std::task::Poll::Ready(None);
                }
                std::task::Poll::Ready(Some(ResponseEnvelope::ok(
                    rid.clone(),
                    serde_json::json!({"tick": 1}),
                )))
            })
            .map(move |env| {
                let _keep_guard = &guard;
                env
            })
        });
        let registry = registry_with_subscription("events/stream", handler);
        let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
        let dp = Dispatcher::new(registry, provider);
        let conn = Arc::new(CallConnection::new(stub_connection()));

        let request = EventEnvelope::requested(
            "sub-abort-1",
            serde_json::json!({
                "operationId": "/events/stream",
                "input": {},
            }),
        );
        let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
        let (send, _sink) = tokio::io::duplex(8 * 1024);
        let stream = crate::core::types::BiStream::from_joined(recv, send);

        let conn_clone = Arc::clone(&conn);
        let dp_clone = dp.clone();
        let handle = tokio::spawn(async move {
            dp_clone.handle_stream(conn_clone, stream).await;
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        dp.handle_abort(&conn, "sub-abort-1").await;
        assert!(
            !conn.pending().lock().contains("sub-abort-1"),
            "abort removes the pending entry"
        );

        handle.abort();
        let _ = handle.await;
        assert!(
            dropped.load(Ordering::SeqCst),
            "stream future dropped → Drop guard released handler resources"
        );
    }

    // --- 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 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()))
        );
    }

    // --- 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 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"
        );
    }

    // --- Unit 9: abort-cancels-Pub (P-06) ----------------------------------

    #[tokio::test]
    async fn pump_sink_same_stream_abort_drops_handler_future_no_response_written() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc as StdArc;

        let dropped = StdArc::new(AtomicBool::new(false));
        let dropped_clone = StdArc::clone(&dropped);
        let handler: crate::registry::registration::SinkHandler = {
            use crate::registry::registration::make_sink_handler;
            use futures::stream::StreamExt;
            make_sink_handler(move |_input, ctx, mut stream| {
                let flag = StdArc::clone(&dropped_clone);
                struct DropGuard(StdArc<AtomicBool>);
                impl Drop for DropGuard {
                    fn drop(&mut self) {
                        self.0.store(true, Ordering::SeqCst);
                    }
                }
                async move {
                    let _guard = DropGuard(StdArc::clone(&flag));
                    while let Some(item) = stream.next().await {
                        if item.is_err() {
                            break;
                        }
                    }
                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({"done": true}))
                }
            })
        };
        let registry = registry_with_pub("fs/upload", 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-abort-ss-1",
            serde_json::json!({
                "operationId": "/fs/upload",
                "input": {},
            }),
        );
        let c1 = EventEnvelope::published("pub-abort-ss-1", serde_json::json!({"chunk": 1}));
        let abort = EventEnvelope::aborted("pub-abort-ss-1");

        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(&abort));

        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);

        dp.handle_stream(conn, stream).await;

        let frames = read_all_frames(&mut sink).await;
        assert!(
            frames.is_empty(),
            "no response written after same-stream abort (handler was dropped, not awaited)"
        );
        assert!(
            dropped.load(Ordering::SeqCst),
            "handler future was dropped (DropGuard fired)"
        );
    }

    #[tokio::test]
    async fn pump_sink_cross_stream_abort_drops_handler_future_no_response_written() {
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc as StdArc;

        let dropped = StdArc::new(AtomicBool::new(false));
        let dropped_clone = StdArc::clone(&dropped);
        let handler: crate::registry::registration::SinkHandler = {
            use crate::registry::registration::make_sink_handler;
            use futures::stream::StreamExt;
            make_sink_handler(move |_input, ctx, mut stream| {
                let flag = StdArc::clone(&dropped_clone);
                struct DropGuard(StdArc<AtomicBool>);
                impl Drop for DropGuard {
                    fn drop(&mut self) {
                        self.0.store(true, Ordering::SeqCst);
                    }
                }
                async move {
                    let _guard = DropGuard(StdArc::clone(&flag));
                    while let Some(item) = stream.next().await {
                        if item.is_err() {
                            break;
                        }
                    }
                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({"done": true}))
                }
            })
        };
        let registry = registry_with_pub("fs/upload", 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-abort-cs-1",
            serde_json::json!({
                "operationId": "/fs/upload",
                "input": {},
            }),
        );
        let c1 = EventEnvelope::published("pub-abort-cs-1", serde_json::json!({"chunk": 1}));

        let (mut read_half, write_half) = tokio::io::duplex(8 * 1024);
        {
            use tokio::io::AsyncWriteExt;
            read_half.write_all(&encode_frame(&request)).await.unwrap();
            read_half.write_all(&encode_frame(&c1)).await.unwrap();
        }

        let recv = tokio::io::BufReader::new(write_half);
        let (send, mut sink) = tokio::io::duplex(8 * 1024);
        let stream = crate::core::types::BiStream::from_joined(recv, send);

        let conn_clone = Arc::clone(&conn);
        let dp_clone = dp.clone();
        let handle = tokio::spawn(async move {
            dp_clone.handle_stream(conn_clone, stream).await;
        });

        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        dp.handle_abort(&conn, "pub-abort-cs-1").await;
        drop(read_half);

        let frames = read_all_frames(&mut sink).await;
        handle.await.unwrap();

        assert!(
            frames.is_empty(),
            "no response written after cross-stream abort"
        );
        assert!(
            dropped.load(Ordering::SeqCst),
            "handler future was dropped (DropGuard fired) via cross-stream abort signal"
        );
    }

    #[tokio::test]
    async fn pump_sink_abort_cleans_up_in_flight_sink_aborts_map() {
        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-abort-cleanup-1",
            serde_json::json!({
                "operationId": "/fs/upload",
                "input": {},
            }),
        );
        let abort = EventEnvelope::aborted("pub-abort-cleanup-1");

        let mut frame_buf = Vec::new();
        frame_buf.extend_from_slice(&encode_frame(&request));
        frame_buf.extend_from_slice(&encode_frame(&abort));

        let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf));
        let (send, _sink) = tokio::io::duplex(8 * 1024);
        let stream = crate::core::types::BiStream::from_joined(recv, send);

        dp.handle_stream(conn, stream).await;

        assert!(
            dp.in_flight_sink_aborts.lock().is_empty(),
            "in_flight_sink_aborts map is cleaned up after pump_sink exits"
        );
    }

    // --- review 004 F-04: the full-duplex serving loop ---------------------

    /// The F-04 probe: two `serve_single_stream` loops over one duplex
    /// — the exact hub↔consumer shape. Each side holds a
    /// `CallConnection` over its own writer; when side A calls
    /// `echo/run`, its `call.requested` crosses the duplex into side
    /// B's serving loop, which dispatches and writes `call.responded`
    /// back, resolving A's pending through the `EVENT_RESPONDED` arm.
    /// This is the direction `read_single_stream_until_closed` used to
    /// drop.
    #[tokio::test]
    async fn serve_single_stream_self_call_resolves_pending() {
        fn echo_registry() -> Arc<crate::registry::registration::OperationRegistry> {
            let registry = crate::registry::registration::OperationRegistry::new();
            registry
                .register(HandlerRegistration::new(
                    external_spec("echo/run", AccessControl::default()),
                    HandlerKind::Once(make_handler(|input, context| async move {
                        ResponseEnvelope::ok(context.request_id, input)
                    })),
                    OperationProvenance::Local,
                    None,
                    None,
                    Capabilities::new(),
                ))
                .unwrap();
            Arc::new(registry)
        }

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);

        // Side B: the duplex end wrapped as one BiStream (read+write);
        // `split_single_stream` divides it into the frame writer and
        // reader the serving loop consumes.
        let server_bidi = crate::core::types::BiStream::from_bidi(server_end);
        let (server_writer, server_reader) =
            crate::protocol::connection::split_single_stream(server_bidi);
        let dp_server = Dispatcher::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
        let server_call_conn = Arc::new(CallConnection::new_single_stream(
            crate::protocol::sink_empty_connection(),
            Arc::clone(&server_writer),
        ));
        let server_conn_for_loop = Arc::clone(&server_call_conn);
        let _serve_b = tokio::spawn(async move {
            dp_server
                .serve_single_stream(server_conn_for_loop, server_reader, server_writer)
                .await;
        });

        // Side A: the same shape on the other duplex end.
        let client_bidi = crate::core::types::BiStream::from_bidi(client_end);
        let (client_writer, client_reader) =
            crate::protocol::connection::split_single_stream(client_bidi);
        let dp_client = Dispatcher::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
        let client_call_conn = Arc::new(CallConnection::new_single_stream(
            crate::protocol::sink_empty_connection(),
            Arc::clone(&client_writer),
        ));
        let client_conn_for_loop = Arc::clone(&client_call_conn);
        let _serve_a = tokio::spawn(async move {
            dp_client
                .serve_single_stream(client_conn_for_loop, client_reader, client_writer)
                .await;
        });

        // Side A calls `echo/run` — side B serves it.
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client_call_conn.call("echo/run", serde_json::json!({ "from": "a" })),
        )
        .await
        .expect("A→B call through serving loops timed out");

        assert!(
            response.result.is_ok(),
            "A→B call resolves, got {:?}",
            response.result
        );
        assert_eq!(response.result.unwrap(), serde_json::json!({ "from": "a" }));

        // And B calls back — A serves it (the both-directions gate).
        let response = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            server_call_conn.call("echo/run", serde_json::json!({ "from": "b" })),
        )
        .await
        .expect("B→A call through serving loops timed out");

        assert!(
            response.result.is_ok(),
            "B→A call resolves, got {:?}",
            response.result
        );
        assert_eq!(response.result.unwrap(), serde_json::json!({ "from": "b" }));
    }
}