alkcall/src/protocol/connection.rs

//! `CallConnection`: an established `alk/call` connection (either
//! direction — accepted or opened). Holds the connection's Layer 2 overlay
//! (imported ops).
//!
//! See `docs/architecture/` for the full specification.

use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use crate::core::auth::Identity;
use crate::core::types::Connection;
use futures::stream::{Stream, StreamExt};
use parking_lot::{Mutex, RwLock};
use serde_json::Value;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::mpsc;

use super::pending::PendingRequestMap;
use super::wire::{
    CallError, EventEnvelope, FrameError, FrameFramedReader, FrameFramedWriter, EVENT_ABORTED,
    EVENT_COMPLETED, EVENT_ERROR, EVENT_RESPONDED,
};
use crate::protocol::wire::ResponseEnvelope;
use crate::registry::context::{generate_request_id, AbortPolicy, OperationContext, ScopedPeerEnv};
use crate::registry::env::OperationEnv;
use crate::registry::registration::{HandlerKind, HandlerRegistration};
use crate::registry::spec::AccessResult;

const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(30);

/// A shared, thread-safe `EventEnvelope` frame writer for single-stream
/// call mode (ADR-036 amendment — channel 0 single-stream mode). Wraps a
/// write half in a `tokio::sync::Mutex` so multiple concurrent calls
/// (`call_with_payload`, `publish_with_payload`, `abort`) can serialize
/// frames on the one `BiStream` without interleaving.
///
/// Each `write_frame` call is atomic under the mutex — the 4-byte length
/// prefix, the JSON body, and the `flush` all happen while the lock is
/// held, so a frame from one call is never partially written when
/// another call's frame begins. This is the single-stream analogue of
/// the stream-per-request model's "one `open_bi` per call" (which
/// naturally serializes per-call writes).
pub struct SharedFrameWriter {
    inner: tokio::sync::Mutex<FrameFramedWriter<Box<dyn AsyncWrite + Send + Unpin>>>,
}

impl SharedFrameWriter {
    pub fn new(writer: Box<dyn AsyncWrite + Send + Unpin>) -> Self {
        Self {
            inner: tokio::sync::Mutex::new(FrameFramedWriter::new(writer)),
        }
    }

    pub async fn write_frame(&self, envelope: &EventEnvelope) -> Result<(), FrameError> {
        self.inner.lock().await.write_frame(envelope).await
    }
}

pub struct CallConnection {
    connection: Option<Arc<Connection>>,
    stored_identity: Option<Identity>,
    imported_operations: Arc<RwLock<HashMap<String, HandlerRegistration>>>,
    pending: Arc<Mutex<PendingRequestMap>>,
    single_stream_writer: Option<Arc<SharedFrameWriter>>,
}

impl Clone for CallConnection {
    fn clone(&self) -> Self {
        Self {
            connection: self.connection.clone(),
            stored_identity: self.stored_identity.clone(),
            imported_operations: Arc::clone(&self.imported_operations),
            pending: Arc::clone(&self.pending),
            single_stream_writer: self.single_stream_writer.clone(),
        }
    }
}

impl CallConnection {
    pub fn new(connection: Connection) -> Self {
        Self {
            connection: Some(Arc::new(connection)),
            stored_identity: None,
            imported_operations: Arc::new(RwLock::new(HashMap::new())),
            pending: Arc::new(Mutex::new(PendingRequestMap::new())),
            single_stream_writer: None,
        }
    }

    pub fn new_overlay_only(identity: Identity) -> Self {
        Self {
            connection: None,
            stored_identity: Some(identity),
            imported_operations: Arc::new(RwLock::new(HashMap::new())),
            pending: Arc::new(Mutex::new(PendingRequestMap::new())),
            single_stream_writer: None,
        }
    }

    /// Construct a `CallConnection` in **single-stream call mode**
    /// (ADR-036 amendment — channel 0 single-stream mode). Used by
    /// `ChannelClient` (the connect side of a channels connection):
    /// channel 0's `BiStream` is the one stream all `EventEnvelope`s
    /// are multiplexed on — `open_bi` is never called (which would
    /// return `StreamClosed` on channel 0's yield-once
    /// `ChannelBidiStreamSource`, ADR-038).
    ///
    /// The caller splits channel 0's `BiStream` and hands the write
    /// half here; the read half is pumped separately (the channels
    /// layer's demux feeds it — channel 0's reassembly buffer is the
    /// read side — so the client does NOT spawn its own read pump on
    /// the single stream; the demux loop routes channel-0 chunks into
    /// the `PendingRequestMap` via `dispatch_envelope`).
    ///
    /// `connection` is the channels-layer `Connection` (carrying
    /// ALPN/addr/identity); `single_stream_writer` is channel 0's
    /// framed write half.
    pub fn new_single_stream(
        connection: Connection,
        single_stream_writer: Arc<SharedFrameWriter>,
    ) -> Self {
        Self {
            connection: Some(Arc::new(connection)),
            stored_identity: None,
            imported_operations: Arc::new(RwLock::new(HashMap::new())),
            pending: Arc::new(Mutex::new(PendingRequestMap::new())),
            single_stream_writer: Some(single_stream_writer),
        }
    }

    /// `true` if this `CallConnection` is in single-stream call mode
    /// (channel 0 multiplexed framing, no `open_bi` per call). The
    /// dispatch loop and the client methods branch on this.
    pub fn is_single_stream(&self) -> bool {
        self.single_stream_writer.is_some()
    }

    /// The shared frame writer for single-stream mode. `None` in
    /// stream-per-request mode.
    pub fn single_stream_writer(&self) -> Option<&Arc<SharedFrameWriter>> {
        self.single_stream_writer.as_ref()
    }

    pub fn connection(&self) -> Option<&Arc<Connection>> {
        self.connection.as_ref()
    }

    pub fn identity(&self) -> Option<&Identity> {
        match &self.connection {
            Some(c) => c.identity(),
            None => self.stored_identity.as_ref(),
        }
    }

    pub fn pending(&self) -> &Arc<Mutex<PendingRequestMap>> {
        &self.pending
    }

    /// Land a peer-announced/discovered bundle on the connection
    /// overlay. Marked (`channel_open`) specs refuse here: a marked op
    /// is unreachable by composition (`OperationEnv::invoke` is
    /// request/response-only; ADR-046 §6) and its data plane has no
    /// overlay path — registration-time loudness, never an advertised
    /// dead body (ADR-047 amendment 4 §4.8; ADR-022's 2026-10-13
    /// amendment). Relay composition
    /// (`ChannelRelay::register_relay_openable`) is the sanctioned
    /// marked-op import.
    pub fn register_imported(&self, registration: HandlerRegistration) -> Result<(), String> {
        if registration.spec.channel_open.is_some() {
            return Err(format!(
                "register_imported: spec `{}` carries the `channel_open` marker — marked \
                 specs are not overlay-importable; compose `ChannelRelay` or consume via \
                 `ChannelClient` (ADR-047 amendment 4 §4.8)",
                registration.spec.name
            ));
        }
        let name = registration.spec.name.clone();
        self.imported_operations.write().insert(name, registration);
        Ok(())
    }

    /// `true` when the connection overlay holds a registration for
    /// `name` — the collision check `op/register`'s replace semantics
    /// need (review 004 F-05).
    pub fn overlay_contains(&self, name: &str) -> bool {
        self.imported_operations.read().contains_key(name)
    }

    /// The connection overlay's registration for `name`, if present.
    pub fn overlay_registration(&self, name: &str) -> Option<HandlerRegistration> {
        self.imported_operations.read().get(name).cloned()
    }

    /// Install a bundle of remote announcements on the connection
    /// overlay. Atomic: if any registration carries the `channel_open`
    /// marker, the overlay is left untouched (nothing partially
    /// installed).
    pub fn register_imported_all(
        &self,
        registrations: Vec<HandlerRegistration>,
    ) -> Result<(), String> {
        let mut overlay = self.imported_operations.write();
        for reg in &registrations {
            if reg.spec.channel_open.is_some() {
                return Err(format!(
                    "register_imported_all: spec `{}` carries the `channel_open` marker — \
                     marked specs are not overlay-importable; compose `ChannelRelay` or \
                     consume via `ChannelClient` (ADR-047 amendment 4 §4.8)",
                    reg.spec.name
                ));
            }
        }
        for reg in registrations {
            let name = reg.spec.name.clone();
            overlay.insert(name, reg);
        }
        Ok(())
    }

    pub fn overlay_env(&self) -> Arc<dyn OperationEnv + Send + Sync> {
        Arc::new(OverlayOperationEnv {
            overlay: Arc::clone(&self.imported_operations),
        })
    }

    pub async fn call(&self, operation_id: &str, input: Value) -> ResponseEnvelope {
        let payload = serde_json::json!({
            "operationId": operation_id,
            "input": input,
        });
        self.call_with_payload(payload).await
    }

    /// Invoke a remote op with a caller-constructed `call.requested` payload.
    /// The payload MUST include `operationId` and `input`; the caller may add
    /// `forwarded_for` (ADR-032) and `auth_token` (ADR-017 §7) for the hub
    /// forwarding path used by `from_call`.
    ///
    /// In stream-per-request mode (top-level `alk/call`), opens a fresh
    /// bidi stream per call and pumps the read half in a spawned task. In
    /// single-stream mode (channel 0, ADR-036 amendment), writes
    /// `call.requested` through the shared frame writer and relies on the
    /// channels-layer demux + `dispatch_envelope` to resolve the pending
    /// entry from the response frame.
    pub async fn call_with_payload(&self, payload: Value) -> ResponseEnvelope {
        let request_id = generate_request_id();

        if let Some(writer) = self.single_stream_writer.clone() {
            return self.call_single_stream(writer, request_id, payload).await;
        }

        let connection = match &self.connection {
            Some(c) => c,
            None => {
                return ResponseEnvelope::error(
                    request_id,
                    CallError::internal("no underlying connection (overlay-only)"),
                );
            }
        };

        // `open_bi` returns a `BiStream` (ADR-009); split it into halves for
        // the call protocol's separate write (request) and read (response)
        // pumps. The split is the stdlib idiom; no per-handler wrapper.
        let stream = match connection.open_bi().await {
            Ok(s) => s,
            Err(err) => {
                let call_error = CallError::internal(format!("failed to open stream: {err}"));
                return ResponseEnvelope::error(request_id, call_error);
            }
        };
        let (recv, send) = tokio::io::split(stream);

        let receiver = {
            let mut pending = self.pending.lock();
            pending.register_call(
                request_id.clone(),
                Some(Instant::now() + DEFAULT_CALL_TIMEOUT),
                None,
            )
        };

        if let Err(err) = self.write_request(send, &request_id, payload).await {
            let call_error = CallError::connection_closed(err);
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            return ResponseEnvelope::error(request_id, call_error);
        }

        let pending = Arc::clone(&self.pending);
        tokio::spawn(async move {
            read_stream_until_closed(recv, &pending).await;
        });

        match receiver.await {
            Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value),
            Ok(Err(error)) => ResponseEnvelope::error(request_id, error),
            Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")),
        }
    }

    /// Single-stream call path (ADR-036 amendment). Writes `call.requested`
    /// through the shared frame writer (no `open_bi`); the channels-layer
    /// demux routes the response frame's `call.responded`/`call.error`/
    /// `call.completed` into the `PendingRequestMap` via
    /// `dispatch_envelope`, resolving the `receiver`.
    async fn call_single_stream(
        &self,
        writer: Arc<SharedFrameWriter>,
        request_id: String,
        payload: Value,
    ) -> ResponseEnvelope {
        let receiver = {
            let mut pending = self.pending.lock();
            pending.register_call(
                request_id.clone(),
                Some(Instant::now() + DEFAULT_CALL_TIMEOUT),
                None,
            )
        };

        let envelope = EventEnvelope::requested(&request_id, payload);
        if let Err(err) = writer.write_frame(&envelope).await {
            let call_error =
                CallError::connection_closed(format!("failed to write request frame: {err}"));
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            return ResponseEnvelope::error(request_id, call_error);
        }

        match receiver.await {
            Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value),
            Ok(Err(error)) => ResponseEnvelope::error(request_id, error),
            Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")),
        }
    }

    pub async fn subscribe(
        &self,
        operation_id: &str,
        input: Value,
    ) -> impl Stream<Item = ResponseEnvelope> {
        let payload = serde_json::json!({
            "operationId": operation_id,
            "input": input,
        });
        self.subscribe_with_payload(payload).await
    }

    /// Subscribe to a remote op with a caller-constructed `call.requested`
    /// payload. The payload MUST include `operationId` and `input`; the
    /// caller may add `forwarded_for` (ADR-032) and `auth_token` (ADR-017 §7)
    /// for the hub forwarding path used by `from_call`'s streaming forwarding
    /// handler. Mirrors [`call_with_payload`](Self::call_with_payload) so the
    /// forwarding handler can populate `forwarded_for` + `auth_token` on the
    /// subscription payload (the plain [`subscribe`](Self::subscribe) builds
    /// the payload internally and omits those fields).
    ///
    /// In single-stream mode (channel 0, ADR-036 amendment), writes
    /// `call.requested` through the shared frame writer; the
    /// channels-layer demux routes `call.responded`/`call.error`/
    /// `call.completed` into the `PendingRequestMap`'s subscription
    /// channel via `dispatch_envelope`.
    pub async fn subscribe_with_payload(
        &self,
        payload: Value,
    ) -> impl Stream<Item = ResponseEnvelope> {
        let request_id = generate_request_id();

        if let Some(writer) = self.single_stream_writer.clone() {
            return self
                .subscribe_single_stream(writer, request_id, payload)
                .await;
        }

        let connection = match &self.connection {
            Some(c) => c,
            None => {
                let call_error = CallError::internal("no underlying connection (overlay-only)");
                return SubscriptionStream::closed(request_id, call_error);
            }
        };

        // `open_bi` returns a `BiStream` (ADR-009); split for the separate
        // write (request) and read (subscription events) pumps.
        let stream = match connection.open_bi().await {
            Ok(s) => s,
            Err(err) => {
                let call_error = CallError::internal(format!("failed to open stream: {err}"));
                return SubscriptionStream::closed(request_id, call_error);
            }
        };
        let (recv, send) = tokio::io::split(stream);

        let receiver = {
            let mut pending = self.pending.lock();
            pending.register_subscribe(request_id.clone(), None, None)
        };

        if let Err(err) = self.write_request(send, &request_id, payload).await {
            let call_error = CallError::connection_closed(err);
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            return SubscriptionStream::closed(request_id, call_error);
        }

        let pending = Arc::clone(&self.pending);
        tokio::spawn(async move {
            read_stream_until_closed(recv, &pending).await;
        });

        SubscriptionStream::new(request_id, receiver)
    }

    /// Single-stream subscribe path (ADR-036 amendment). Writes
    /// `call.requested` through the shared frame writer; the
    /// channels-layer demux routes events into the subscription
    /// channel via `dispatch_envelope`.
    async fn subscribe_single_stream(
        &self,
        writer: Arc<SharedFrameWriter>,
        request_id: String,
        payload: Value,
    ) -> SubscriptionStream {
        let receiver = {
            let mut pending = self.pending.lock();
            pending.register_subscribe(request_id.clone(), None, None)
        };

        let envelope = EventEnvelope::requested(&request_id, payload);
        if let Err(err) = writer.write_frame(&envelope).await {
            let call_error =
                CallError::connection_closed(format!("failed to write request frame: {err}"));
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            return SubscriptionStream::closed(request_id, call_error);
        }

        SubscriptionStream::new(request_id, receiver)
    }

    pub async fn abort(&self, request_id: &str) {
        let envelope = EventEnvelope::aborted(request_id);

        if let Some(writer) = self.single_stream_writer.as_ref() {
            if let Err(err) = writer.write_frame(&envelope).await {
                tracing::warn!(error = %err, request_id, "failed to send call.aborted");
                return;
            }
            self.pending.lock().handle_aborted(request_id);
            return;
        }

        if let Err(err) = self.write_envelope(&envelope).await {
            tracing::warn!(error = %err, request_id, "failed to send call.aborted");
            return;
        }
        self.pending.lock().handle_aborted(request_id);
    }

    /// Publish a stream to a remote `Pub` operation (ADR-046). Sends
    /// `call.requested` (open), then one `call.published` per item in
    /// `stream`, then `call.completed` (stream end). Returns the
    /// responder's single `ResponseEnvelope`.
    ///
    /// The publish is long-lived and unbounded — no client-side deadline
    /// (ADR-046 §7, "same as `Sub`"). The `PendingRequestMap` entry is
    /// registered with `timeout: None`, so the sweeper never evicts it.
    pub async fn publish(
        &self,
        operation_id: &str,
        input: Value,
        stream: Pin<Box<dyn Stream<Item = Value> + Send>>,
    ) -> ResponseEnvelope {
        let payload = serde_json::json!({
            "operationId": operation_id,
            "input": input,
        });
        self.publish_with_payload(payload, stream).await
    }

    /// Publish a stream to a remote `Pub` op with a caller-constructed
    /// `call.requested` payload. The payload MUST include `operationId`
    /// and `input`; the caller may add `forwarded_for` (ADR-032) and
    /// `auth_token` for the hub forwarding path used by `from_call`'s
    /// sink forwarding handler. Mirrors
    /// [`subscribe_with_payload`](Self::subscribe_with_payload).
    ///
    /// In stream-per-request mode (top-level `alk/call`), opens a
    /// fresh bidi stream, writes `call.requested` + chunks +
    /// `call.completed` on the same write half, and pumps the read half
    /// concurrently for the single `call.responded` / `call.error`
    /// response (ADR-046 §5 stream lifecycle).
    ///
    /// In single-stream mode (channel 0, ADR-036 amendment), writes
    /// `call.requested` + chunks + `call.completed` through the shared
    /// frame writer; the channels-layer demux routes the responder's
    /// `call.responded` / `call.error` into the `PendingRequestMap`
    /// via `dispatch_envelope`.
    pub async fn publish_with_payload(
        &self,
        payload: Value,
        stream: Pin<Box<dyn Stream<Item = Value> + Send>>,
    ) -> ResponseEnvelope {
        let request_id = generate_request_id();

        if let Some(writer) = self.single_stream_writer.clone() {
            return self
                .publish_single_stream(writer, request_id, payload, stream)
                .await;
        }

        let connection = match &self.connection {
            Some(c) => c,
            None => {
                return ResponseEnvelope::error(
                    request_id,
                    CallError::internal("no underlying connection (overlay-only)"),
                );
            }
        };

        let stream_a = match connection.open_bi().await {
            Ok(s) => s,
            Err(err) => {
                return ResponseEnvelope::error(
                    request_id,
                    CallError::internal(format!("failed to open stream: {err}")),
                );
            }
        };
        let (recv, send) = tokio::io::split(stream_a);

        let receiver = {
            let mut pending = self.pending.lock();
            pending.register_call(request_id.clone(), None, None)
        };

        let pending_map = Arc::clone(&self.pending);
        let read_handle = tokio::spawn(async move {
            read_stream_until_closed(recv, &pending_map).await;
        });

        let write_result = pump_publish_to_wire(send, &request_id, payload, stream).await;
        if let Err((stage, err)) = write_result {
            // A write failure before the request frame could be delivered is
            // provably undelivered — retry is safe. Mid-publish failures leave
            // delivery ambiguous; the INTERNAL mapping (non-retryable) is the
            // safe default there.
            let call_error = match stage {
                PublishWriteStage::Request => CallError::connection_closed(err),
                PublishWriteStage::Published | PublishWriteStage::Completed => {
                    CallError::internal(err)
                }
            };
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            read_handle.abort();
            return ResponseEnvelope::error(request_id, call_error);
        }

        match receiver.await {
            Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value),
            Ok(Err(error)) => ResponseEnvelope::error(request_id, error),
            Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")),
        }
    }

    /// Single-stream publish path (ADR-036 amendment). Pumps
    /// `call.requested` + `call.published` chunks + `call.completed`
    /// through the shared frame writer; the channels-layer demux routes
    /// the responder's `call.responded` / `call.error` into the
    /// `PendingRequestMap` via `dispatch_envelope`.
    async fn publish_single_stream(
        &self,
        writer: Arc<SharedFrameWriter>,
        request_id: String,
        payload: Value,
        mut stream: Pin<Box<dyn Stream<Item = Value> + Send>>,
    ) -> ResponseEnvelope {
        let receiver = {
            let mut pending = self.pending.lock();
            pending.register_call(request_id.clone(), None, None)
        };

        let requested = EventEnvelope::requested(&request_id, payload);
        if let Err(err) = writer.write_frame(&requested).await {
            let call_error =
                CallError::connection_closed(format!("failed to write request frame: {err}"));
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            return ResponseEnvelope::error(request_id, call_error);
        }
        while let Some(chunk) = stream.next().await {
            let envelope = EventEnvelope::published(&request_id, chunk);
            if let Err(err) = writer.write_frame(&envelope).await {
                let call_error =
                    CallError::internal(format!("failed to write published frame: {err}"));
                self.pending
                    .lock()
                    .handle_error(&request_id, call_error.clone());
                return ResponseEnvelope::error(request_id, call_error);
            }
        }
        let completed = EventEnvelope::completed(&request_id);
        if let Err(err) = writer.write_frame(&completed).await {
            let call_error = CallError::internal(format!("failed to write completed frame: {err}"));
            self.pending
                .lock()
                .handle_error(&request_id, call_error.clone());
            return ResponseEnvelope::error(request_id, call_error);
        }

        match receiver.await {
            Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value),
            Ok(Err(error)) => ResponseEnvelope::error(request_id, error),
            Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")),
        }
    }

    async fn write_request<W>(
        &self,
        send: W,
        request_id: &str,
        payload: Value,
    ) -> Result<(), String>
    where
        W: AsyncWrite + Unpin,
    {
        let envelope = EventEnvelope::requested(request_id, payload);
        let mut writer = FrameFramedWriter::new(send);
        writer
            .write_frame(&envelope)
            .await
            .map_err(|e| format!("failed to write frame: {e}"))
    }

    async fn write_envelope(&self, envelope: &EventEnvelope) -> Result<(), String> {
        let connection = self
            .connection
            .as_ref()
            .ok_or_else(|| "no underlying connection (overlay-only)".to_string())?;
        // `open_bi` returns a `BiStream` (ADR-009). We only need the write
        // half to send the envelope; split and drop the read half.
        let stream = connection
            .open_bi()
            .await
            .map_err(|e| format!("failed to open stream: {e}"))?;
        let (_recv, send) = tokio::io::split(stream);
        let mut writer = FrameFramedWriter::new(send);
        writer
            .write_frame(envelope)
            .await
            .map_err(|e| format!("failed to write frame: {e}"))
    }
}

/// Which frame of a publish pump failed to write. The `Request` stage is
/// provably undelivered (the producer never received `call.requested`, so a
/// reconnect/retry is safe); the later stages leave delivery ambiguous.
enum PublishWriteStage {
    Request,
    Published,
    Completed,
}

async fn pump_publish_to_wire<W>(
    send: W,
    request_id: &str,
    payload: Value,
    mut stream: Pin<Box<dyn Stream<Item = Value> + Send>>,
) -> Result<(), (PublishWriteStage, String)>
where
    W: AsyncWrite + Unpin,
{
    let mut writer = FrameFramedWriter::new(send);
    let requested = EventEnvelope::requested(request_id, payload);
    writer.write_frame(&requested).await.map_err(|e| {
        (
            PublishWriteStage::Request,
            format!("failed to write request frame: {e}"),
        )
    })?;
    while let Some(chunk) = stream.next().await {
        let envelope = EventEnvelope::published(request_id, chunk);
        writer.write_frame(&envelope).await.map_err(|e| {
            (
                PublishWriteStage::Published,
                format!("failed to write published frame: {e}"),
            )
        })?;
    }
    let completed = EventEnvelope::completed(request_id);
    writer.write_frame(&completed).await.map_err(|e| {
        (
            PublishWriteStage::Completed,
            format!("failed to write completed frame: {e}"),
        )
    })?;
    Ok(())
}

async fn read_stream_until_closed<R>(recv: R, pending: &Arc<Mutex<PendingRequestMap>>)
where
    R: AsyncRead + Unpin,
{
    let mut reader = FrameFramedReader::new(recv);
    while let Ok(envelope) = reader.read_frame().await {
        dispatch_envelope(pending, envelope);
    }
}

/// Read `EventEnvelope` frames off a boxed read half and route them
/// into the `PendingRequestMap` via `dispatch_envelope` until the read
/// half closes. The single-stream analogue of `read_stream_until_closed`
/// (ADR-036 amendment — channel 0 single-stream mode). Used by
/// `ChannelClient::from_connection` to drive the client-side response
/// pump: the channels-layer demux feeds channel-0 chunks into the
/// reassembled read half, this pump reads `EventEnvelope` frames off
/// it, and `dispatch_envelope` resolves pending calls.
pub async fn read_single_stream_until_closed(
    reader: Box<dyn AsyncRead + Send + Unpin>,
    pending: &Arc<Mutex<PendingRequestMap>>,
) {
    read_stream_until_closed(reader, pending).await;
}

/// Split a `BiStream` into a `SharedFrameWriter` (the framed write
/// half, wrapped in a `tokio::sync::Mutex` for concurrent frame writes)
/// and the raw read half (for the single-stream dispatch loop to wrap
/// in a `FrameFramedReader`). Used by single-stream call mode
/// (ADR-036 amendment) on both the connect side (`ChannelClient`) and
/// the accept side (`ChannelsAdapter`'s `install_channel_zero` hook):
/// channel 0's `BiStream` is the one stream all `EventEnvelope`s are
/// multiplexed on.
pub fn split_single_stream(
    stream: crate::core::types::BiStream,
) -> (Arc<SharedFrameWriter>, Box<dyn AsyncRead + Send + Unpin>) {
    let (recv, send) = tokio::io::split(stream);
    let writer = Arc::new(SharedFrameWriter::new(Box::new(send)));
    let reader: Box<dyn AsyncRead + Send + Unpin> = Box::new(recv);
    (writer, reader)
}

fn dispatch_envelope(pending: &Arc<Mutex<PendingRequestMap>>, envelope: EventEnvelope) {
    let request_id = envelope.id.clone();
    match envelope.r#type.as_str() {
        EVENT_RESPONDED => {
            let output = envelope
                .payload
                .get("output")
                .cloned()
                .unwrap_or(Value::Null);
            pending.lock().handle_responded(&request_id, output);
        }
        EVENT_COMPLETED => {
            pending.lock().handle_completed(&request_id);
        }
        EVENT_ABORTED => {
            pending.lock().handle_aborted(&request_id);
        }
        EVENT_ERROR => {
            if let Ok(error) = serde_json::from_value::<CallError>(envelope.payload) {
                pending.lock().handle_error(&request_id, error);
            }
        }
        _ => {}
    }
}

struct OverlayOperationEnv {
    overlay: Arc<RwLock<HashMap<String, HandlerRegistration>>>,
}

#[async_trait::async_trait]
impl OperationEnv for OverlayOperationEnv {
    async fn invoke_with_policy(
        &self,
        namespace: &str,
        operation: &str,
        input: Value,
        parent: &OperationContext,
        policy: AbortPolicy,
    ) -> ResponseEnvelope {
        let name = format!("{namespace}/{operation}");

        if !parent.scoped_env.allows(&name) {
            return ResponseEnvelope::not_found(parent.request_id.clone(), &name);
        }

        let handler: HandlerKind;
        let composition_authority;
        let scoped_env;
        let access_control;
        let resource_id_path;
        {
            let overlay = self.overlay.read();
            let Some(registration) = overlay.get(&name) else {
                return ResponseEnvelope::not_found(parent.request_id.clone(), &name);
            };
            handler = registration.handler.clone();
            composition_authority = registration.composition_authority.clone();
            scoped_env = registration
                .scoped_env
                .clone()
                .unwrap_or_else(ScopedPeerEnv::empty);
            access_control = registration.spec.access_control.clone();
            resource_id_path = registration.spec.resource_id_path.clone();
        }

        let caller_identity = if parent.internal {
            parent
                .handler_identity
                .as_ref()
                .and_then(|ca| ca.as_identity())
        } else {
            parent.identity.clone()
        };
        let resource_id = resource_id_path
            .as_ref()
            .and_then(|path| crate::registry::registration::extract_json_pointer(&input, path));
        if let AccessResult::Forbidden(message) = access_control.check(
            caller_identity.as_ref(),
            resource_id.as_deref(),
            parent.ownership.as_deref(),
        ) {
            return ResponseEnvelope::forbidden(parent.request_id.clone(), message);
        }

        let context = OperationContext {
            request_id: generate_request_id(),
            parent_request_id: Some(parent.request_id.clone()),
            identity: parent
                .handler_identity
                .as_ref()
                .and_then(|ca| ca.as_identity()),
            handler_identity: composition_authority,
            forwarded_for: None,
            capabilities: parent.capabilities.clone(),
            metadata: HashMap::new(),
            abort_policy: policy,
            deadline: parent.deadline,
            scoped_env,
            env: parent.env.clone(),
            internal: true,
            ownership: parent.ownership.clone(),
        };

        match handler {
            HandlerKind::Once(h) => h(input, context).await,
            HandlerKind::Stream(_) => ResponseEnvelope::error(
                parent.request_id.clone(),
                CallError::invalid_operation_type(
                    "OperationEnv::invoke() called on a Sub op; composition is request/response-only",
                ),
            ),
            HandlerKind::Sink(_) => ResponseEnvelope::error(
                parent.request_id.clone(),
                CallError::invalid_operation_type(
                    "OperationEnv::invoke() called on a Pub op; composition is request/response-only",
                ),
            ),
        }
    }

    fn contains(&self, name: &str) -> bool {
        self.overlay.read().contains_key(name)
    }

    fn list_operation_names(&self) -> Vec<String> {
        self.overlay.read().keys().cloned().collect()
    }
}

pub struct SubscriptionStream {
    request_id: String,
    receiver: mpsc::Receiver<Result<Value, CallError>>,
    done: bool,
}

impl SubscriptionStream {
    fn new(request_id: String, receiver: mpsc::Receiver<Result<Value, CallError>>) -> Self {
        Self {
            request_id,
            receiver,
            done: false,
        }
    }

    fn closed(request_id: String, error: CallError) -> Self {
        let (tx, rx) = mpsc::channel(1);
        let _ = tx.try_send(Err(error));
        Self {
            request_id,
            receiver: rx,
            done: false,
        }
    }
}

impl Stream for SubscriptionStream {
    type Item = ResponseEnvelope;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if self.done {
            return Poll::Ready(None);
        }
        let this = self.get_mut();
        match this.receiver.poll_recv(cx) {
            Poll::Ready(None) => {
                this.done = true;
                Poll::Ready(None)
            }
            Poll::Ready(Some(Ok(value))) => {
                Poll::Ready(Some(ResponseEnvelope::ok(this.request_id.clone(), value)))
            }
            Poll::Ready(Some(Err(error))) => {
                this.done = true;
                Poll::Ready(Some(ResponseEnvelope::error(
                    this.request_id.clone(),
                    error,
                )))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::types::Capabilities;
    use crate::protocol::wire::{
        EVENT_ABORTED, EVENT_COMPLETED, EVENT_ERROR, EVENT_PUBLISHED, EVENT_REQUESTED,
    };
    use crate::registry::context::CompositionAuthority;
    use crate::registry::registration::{
        make_handler, Handler, HandlerKind, HandlerRegistration, OperationProvenance,
        OperationRegistry,
    };
    use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
    use std::collections::HashMap;
    use std::time::{Duration, Instant};

    use crate::protocol::sink_empty_connection as stub_connection;

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

    fn echo_handler() -> Handler {
        make_handler(
            |input, context| async move { ResponseEnvelope::ok(context.request_id, input) },
        )
    }

    fn imported_registration(name: &str) -> HandlerRegistration {
        HandlerRegistration::new(
            external_spec(name),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        )
    }

    fn root_context(
        request_id: &str,
        scoped_env: ScopedPeerEnv,
        env: Arc<dyn OperationEnv + Send + Sync>,
    ) -> OperationContext {
        OperationContext {
            request_id: request_id.to_string(),
            parent_request_id: None,
            identity: None,
            handler_identity: Some(CompositionAuthority::new("agent", ["fs:read".to_string()])),
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env,
            env,
            abort_policy: AbortPolicy::default(),
            deadline: Some(Instant::now() + Duration::from_secs(30)),
            internal: true,
            ownership: None,
        }
    }

    #[test]
    fn register_imported_adds_to_overlay_and_contains_returns_true() {
        let conn = CallConnection::new(stub_connection());
        let env = conn.overlay_env();

        assert!(!env.contains("worker/exec"));

        let _error = conn.register_imported(imported_registration("worker/exec"));

        assert!(env.contains("worker/exec"));
        assert!(!env.contains("worker/missing"));
    }

    #[test]
    fn register_imported_all_bulk_adds_to_overlay() {
        let conn = CallConnection::new(stub_connection());
        let env = conn.overlay_env();

        let registrations_result = conn.register_imported_all(vec![
            imported_registration("worker/exec"),
            imported_registration("worker/status"),
            imported_registration("fs/readFile"),
        ]);
        let _error = registrations_result;

        assert!(env.contains("worker/exec"));
        assert!(env.contains("worker/status"));
        assert!(env.contains("fs/readFile"));
        assert!(!env.contains("worker/missing"));
    }

    #[tokio::test]
    async fn overlay_env_dispatches_to_imported_op() {
        let conn = CallConnection::new(stub_connection());
        let _error = conn.register_imported(imported_registration("worker/exec"));
        let env = conn.overlay_env();

        let scoped = ScopedPeerEnv::new(["worker/exec"]);
        let ctx = root_context("root-1", scoped, env.clone());

        let response = env
            .invoke("worker", "exec", serde_json::json!({"hi": 1}), &ctx)
            .await;

        assert!(response.result.is_ok());
        assert_eq!(response.result.unwrap(), serde_json::json!({"hi": 1}));
    }

    #[tokio::test]
    async fn overlay_env_contains_returns_false_for_non_imported_op() {
        let conn = CallConnection::new(stub_connection());
        let _error = conn.register_imported(imported_registration("worker/exec"));
        let env = conn.overlay_env();

        assert!(!env.contains("worker/missing"));

        let scoped = ScopedPeerEnv::new(["worker/missing"]);
        let ctx = root_context("root-2", scoped, env.clone());

        let response = env
            .invoke("worker", "missing", serde_json::json!({}), &ctx)
            .await;

        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn overlay_env_reachability_check_returns_not_found_for_disallowed_op() {
        let conn = CallConnection::new(stub_connection());
        let _error = conn.register_imported(imported_registration("worker/exec"));
        let env = conn.overlay_env();

        let scoped = ScopedPeerEnv::empty();
        let ctx = root_context("root-3", scoped, env.clone());

        let response = env
            .invoke("worker", "exec", serde_json::json!({}), &ctx)
            .await;

        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn overlay_env_dispatched_child_has_internal_true_and_parent_set() {
        let conn = CallConnection::new(stub_connection());
        let inspect_handler = make_handler(|_input, context| async move {
            let internal = context.is_internal();
            let parent_set = context.parent_request_id.is_some();
            ResponseEnvelope::ok(
                context.request_id,
                serde_json::json!({
                    "internal": internal,
                    "parent_set": parent_set,
                }),
            )
        });
        let _error = conn.register_imported(HandlerRegistration::new(
            external_spec("worker/exec"),
            HandlerKind::Once(inspect_handler),
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        ));
        let env = conn.overlay_env();

        let scoped = ScopedPeerEnv::new(["worker/exec"]);
        let ctx = root_context("root-4", scoped, env.clone());

        let response = env
            .invoke("worker", "exec", serde_json::json!({}), &ctx)
            .await;
        let out = response.result.expect("ok");
        assert_eq!(out["internal"], Value::Bool(true));
        assert_eq!(out["parent_set"], Value::Bool(true));
    }

    #[test]
    fn connection_accessor_returns_underlying_connection() {
        let conn = CallConnection::new(stub_connection());
        assert_eq!(
            conn.connection()
                .expect("quic connection present")
                .remote_alpn(),
            b"alk/call"
        );
    }

    #[test]
    fn empty_overlay_contains_nothing() {
        let conn = CallConnection::new(stub_connection());
        let env = conn.overlay_env();
        assert!(!env.contains("anything"));
        assert!(!env.contains(""));
    }

    #[test]
    fn overlay_drops_with_connection() {
        let captured: Arc<RwLock<HashMap<String, HandlerRegistration>>> =
            Arc::new(RwLock::new(HashMap::new()));
        {
            let conn = CallConnection::new(stub_connection());
            let _error = conn.register_imported(imported_registration("worker/exec"));
            assert!(conn.overlay_env().contains("worker/exec"));
            std::mem::swap(
                &mut *captured.write(),
                &mut *conn.imported_operations.write(),
            );
        }
        assert!(captured.read().contains_key("worker/exec"));
    }

    // --- dispatch_envelope -------------------------------------------------

    fn empty_pending() -> Arc<Mutex<PendingRequestMap>> {
        Arc::new(Mutex::new(PendingRequestMap::new()))
    }

    #[tokio::test]
    async fn dispatch_envelope_responded_resolves_call_receiver() {
        let pending = empty_pending();
        let rx = pending.lock().register_call(
            "req-1".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        let envelope = EventEnvelope::responded("req-1", serde_json::json!({"v": 42}));
        dispatch_envelope(&pending, envelope);
        assert!(!pending.lock().contains("req-1"));
        let result = tokio::time::timeout(Duration::from_millis(100), rx).await;
        match result {
            Ok(Ok(Ok(value))) => assert_eq!(value, serde_json::json!({"v": 42})),
            other => panic!("expected Ok({{v:42}}), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_envelope_responded_pushes_to_subscribe_channel() {
        let pending = empty_pending();
        let mut rx = pending
            .lock()
            .register_subscribe("sub-1".to_string(), None, None);
        dispatch_envelope(
            &pending,
            EventEnvelope::responded("sub-1", serde_json::json!("first")),
        );
        dispatch_envelope(
            &pending,
            EventEnvelope::responded("sub-1", serde_json::json!("second")),
        );
        assert!(pending.lock().contains("sub-1"));
        let a = tokio::time::timeout(Duration::from_millis(100), rx.recv()).await;
        let b = tokio::time::timeout(Duration::from_millis(100), rx.recv()).await;
        match (a, b) {
            (Ok(Some(Ok(x))), Ok(Some(Ok(y)))) => {
                assert_eq!(x, serde_json::json!("first"));
                assert_eq!(y, serde_json::json!("second"));
            }
            other => panic!("expected two Ok values, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_envelope_completed_removes_entry() {
        let pending = empty_pending();
        let _rx = pending
            .lock()
            .register_subscribe("sub-2".to_string(), None, None);
        assert!(pending.lock().contains("sub-2"));
        dispatch_envelope(&pending, EventEnvelope::completed("sub-2"));
        assert!(!pending.lock().contains("sub-2"));
    }

    #[tokio::test]
    async fn dispatch_envelope_aborted_removes_entry() {
        let pending = empty_pending();
        let _rx = pending.lock().register_call(
            "req-2".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        assert!(pending.lock().contains("req-2"));
        dispatch_envelope(&pending, EventEnvelope::aborted("req-2"));
        assert!(!pending.lock().contains("req-2"));
    }

    #[tokio::test]
    async fn dispatch_envelope_error_resolves_call_with_error() {
        let pending = empty_pending();
        let rx = pending.lock().register_call(
            "req-3".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        let err = CallError::new("FILE_NOT_FOUND", "missing", false);
        dispatch_envelope(&pending, EventEnvelope::error("req-3", &err));
        assert!(!pending.lock().contains("req-3"));
        let result = tokio::time::timeout(Duration::from_millis(100), rx).await;
        match result {
            Ok(Ok(Err(e))) => {
                assert_eq!(e.code, "FILE_NOT_FOUND");
                assert!(!e.retryable);
            }
            other => panic!("expected Err(FILE_NOT_FOUND), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_envelope_error_pushes_error_to_subscribe_channel() {
        let pending = empty_pending();
        let mut rx = pending
            .lock()
            .register_subscribe("sub-3".to_string(), None, None);
        let err = CallError::new("RATE_LIMITED", "slow down", true);
        dispatch_envelope(&pending, EventEnvelope::error("sub-3", &err));
        assert!(!pending.lock().contains("sub-3"));
        let result = tokio::time::timeout(Duration::from_millis(100), rx.recv()).await;
        match result {
            Ok(Some(Err(e))) => {
                assert_eq!(e.code, "RATE_LIMITED");
                assert!(e.retryable);
            }
            other => panic!("expected Err(RATE_LIMITED), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn dispatch_envelope_error_with_invalid_payload_is_no_op() {
        let pending = empty_pending();
        let _rx = pending.lock().register_call(
            "req-4".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        let malformed =
            EventEnvelope::new(EVENT_ERROR, "req-4", serde_json::json!("not-an-object"));
        dispatch_envelope(&pending, malformed);
        assert!(pending.lock().contains("req-4"));
    }

    #[tokio::test]
    async fn dispatch_envelope_unknown_event_type_is_no_op() {
        let pending = empty_pending();
        let _rx = pending.lock().register_call(
            "req-5".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        let unknown = EventEnvelope::new("call.mystery", "req-5", serde_json::json!({}));
        dispatch_envelope(&pending, unknown);
        assert!(pending.lock().contains("req-5"));
    }

    #[tokio::test]
    async fn dispatch_envelope_unknown_request_id_is_no_op() {
        let pending = empty_pending();
        dispatch_envelope(
            &pending,
            EventEnvelope::responded("ghost", serde_json::json!(1)),
        );
        dispatch_envelope(&pending, EventEnvelope::completed("ghost"));
        dispatch_envelope(&pending, EventEnvelope::aborted("ghost"));
        assert!(pending.lock().is_empty());
    }

    // --- SubscriptionStream ------------------------------------------------

    #[tokio::test]
    async fn subscription_stream_closed_yields_one_error_then_ends() {
        use futures::stream::StreamExt;
        let err = CallError::internal("stream closed before send");
        let mut stream = SubscriptionStream::closed("req-x".to_string(), err);
        let first = stream.next().await;
        match first {
            Some(env) => {
                assert_eq!(env.request_id, "req-x");
                assert!(env.result.is_err());
                assert_eq!(env.result.unwrap_err().code, "INTERNAL");
            }
            other => panic!("expected one error envelope, got {other:?}"),
        }
        let second = stream.next().await;
        assert!(second.is_none(), "stream must terminate after the error");
    }

    #[tokio::test]
    async fn subscription_stream_emits_ok_values_then_completes() {
        use futures::stream::StreamExt;
        let (tx, rx) = mpsc::channel(8);
        let mut stream = SubscriptionStream::new("req-y".to_string(), rx);
        tx.try_send(Ok(serde_json::json!(1))).unwrap();
        tx.try_send(Ok(serde_json::json!(2))).unwrap();
        drop(tx);

        let a = stream.next().await.unwrap();
        assert_eq!(a.request_id, "req-y");
        assert_eq!(a.result.unwrap(), serde_json::json!(1));
        let b = stream.next().await.unwrap();
        assert_eq!(b.result.unwrap(), serde_json::json!(2));
        assert!(
            stream.next().await.is_none(),
            "stream ends after channel closes"
        );
    }

    #[tokio::test]
    async fn subscription_stream_emits_error_then_terminates() {
        use futures::stream::StreamExt;
        let (tx, rx) = mpsc::channel(8);
        let mut stream = SubscriptionStream::new("req-z".to_string(), rx);
        tx.try_send(Ok(serde_json::json!("ok"))).unwrap();
        tx.try_send(Err(CallError::timeout("timed out"))).unwrap();
        drop(tx);

        let first = stream.next().await.unwrap();
        assert_eq!(first.result.unwrap(), serde_json::json!("ok"));
        let second = stream.next().await.unwrap();
        assert_eq!(second.request_id, "req-z");
        assert_eq!(second.result.unwrap_err().code, "TIMEOUT");
        assert!(
            stream.next().await.is_none(),
            "stream terminates after error"
        );
    }

    // --- non-QUIC (overlay-only) CallConnection ---------------------------

    fn sample_identity(id: &str) -> Identity {
        Identity {
            id: id.to_string(),
            scopes: vec!["fs:read".to_string()],
            resources: HashMap::new(),
        }
    }

    #[test]
    fn overlay_only_constructor_has_no_quic_connection() {
        let conn = CallConnection::new_overlay_only(sample_identity("ws-peer"));
        assert!(conn.connection().is_none(), "no QUIC connection stored");
    }

    #[test]
    fn overlay_only_identity_returns_stored_identity() {
        let conn = CallConnection::new_overlay_only(sample_identity("ws-peer"));
        let identity = conn.identity().expect("identity stored");
        assert_eq!(identity.id, "ws-peer");
        assert_eq!(identity.scopes, vec!["fs:read".to_string()]);
    }

    #[test]
    fn overlay_only_holds_independent_pending_map() {
        let conn = CallConnection::new_overlay_only(sample_identity("ws-peer"));
        let pending = Arc::clone(conn.pending());
        assert!(pending.lock().is_empty());
        let _rx = pending.lock().register_call(
            "req-overlay-1".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        assert!(pending.lock().contains("req-overlay-1"));
    }

    #[test]
    fn overlay_only_register_imported_populates_overlay() {
        let conn = CallConnection::new_overlay_only(sample_identity("ws-peer"));
        let env = conn.overlay_env();
        assert!(!env.contains("worker/exec"));
        let _error = conn.register_imported(imported_registration("worker/exec"));
        assert!(env.contains("worker/exec"));
    }

    #[tokio::test]
    async fn overlay_only_overlay_env_dispatches_imported_op() {
        let conn = CallConnection::new_overlay_only(sample_identity("ws-peer"));
        let _error = conn.register_imported(imported_registration("worker/exec"));
        let env = conn.overlay_env();

        let scoped = ScopedPeerEnv::new(["worker/exec"]);
        let ctx = root_context("overlay-root-1", scoped, env.clone());

        let response = env
            .invoke("worker", "exec", serde_json::json!({"v": 1}), &ctx)
            .await;
        assert!(response.result.is_ok());
        assert_eq!(response.result.unwrap(), serde_json::json!({"v": 1}));
    }

    #[tokio::test]
    async fn overlay_only_call_without_connection_returns_error() {
        let conn = CallConnection::new_overlay_only(sample_identity("ws-peer"));
        let response = conn.call("fs/readFile", serde_json::json!({})).await;
        let err = response.result.expect_err("no connection → error");
        assert_eq!(err.code, "INTERNAL");
    }

    #[test]
    fn quic_path_identity_returns_connection_identity() {
        let conn = CallConnection::new(stub_connection());
        conn.connection()
            .expect("quic connection present")
            .set_identity(sample_identity("quic-peer"))
            .expect("identity not yet set");
        let identity = conn.identity().expect("identity from connection");
        assert_eq!(identity.id, "quic-peer");
    }

    #[test]
    fn quic_path_stored_identity_is_none_when_connection_present() {
        let conn = CallConnection::new(stub_connection());
        assert!(conn.connection().is_some(), "QUIC connection present");
        assert!(conn.identity().is_none(), "no identity set yet");
    }

    #[tokio::test]
    async fn overlay_env_invoke_on_stream_kind_returns_invalid_operation_type() {
        use crate::registry::registration::make_streaming_handler;
        let conn = CallConnection::new(stub_connection());
        let streaming_handler = make_streaming_handler(|input, ctx| {
            futures::stream::iter(vec![ResponseEnvelope::ok(ctx.request_id, input)])
        });
        let _error = conn.register_imported(HandlerRegistration::new(
            OperationSpec::new(
                "events/stream",
                OperationType::Sub,
                Visibility::External,
                serde_json::json!({}),
                serde_json::json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Stream(streaming_handler),
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        ));
        let env = conn.overlay_env();
        let scoped = ScopedPeerEnv::new(["events/stream"]);
        let ctx = root_context("root-stream", scoped, env.clone());
        let response = env
            .invoke("events", "stream", serde_json::json!({}), &ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    // --- end-to-end publish (P-12 #1, ADR-046 §8) ------------------------
    //
    // Wires `CallConnection::publish` (client) ↔ `Dispatcher::run_loop`
    // (responder) over a real `tokio::io::duplex` pair. This is the
    // acceptance gate for Unit 1: it would have caught P-01 (chunks on
    // the wrong stream) immediately — the responder would never see the
    // chunks.

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

    #[tokio::test]
    async fn publish_end_to_end_delivers_chunks_and_returns_response() {
        use crate::core::auth::IdentityProvider;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::duplex_connection_pair;
        use crate::registry::registration::{make_sink_handler, SinkHandler};
        use futures::stream::StreamExt;

        struct NoopIdProvider;
        impl IdentityProvider for NoopIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
                None
            }
            fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
                None
            }
        }

        let counting_sink: SinkHandler = make_sink_handler(|_input, ctx, mut stream| async move {
            let mut count = 0u32;
            let mut last = serde_json::Value::Null;
            while let Some(item) = stream.next().await {
                match item {
                    Ok(v) => {
                        count += 1;
                        last = v;
                    }
                    Err(_) => break,
                }
            }
            ResponseEnvelope::ok(
                ctx.request_id,
                serde_json::json!({ "count": count, "last": last }),
            )
        });

        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec_e2e("fs/upload"),
                HandlerKind::Sink(counting_sink),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let (client_conn, server_conn) = duplex_connection_pair(64 * 1024);
        let client = CallConnection::new(client_conn);

        let dp = Dispatcher::new(Arc::clone(&registry), Arc::new(NoopIdProvider));
        let server_call_conn = Arc::new(CallConnection::new(server_conn));
        let server_handle = tokio::spawn(async move {
            dp.run_loop(server_call_conn).await;
        });

        let chunks = vec![
            serde_json::json!({"chunk": 1}),
            serde_json::json!({"chunk": 2}),
            serde_json::json!({"chunk": 3}),
        ];
        let stream: Pin<Box<dyn Stream<Item = serde_json::Value> + Send>> =
            Box::pin(futures::stream::iter(chunks.clone()));
        let response = client
            .publish("fs/upload", serde_json::json!({"path": "/x"}), stream)
            .await;

        assert!(
            response.result.is_ok(),
            "publish should succeed, got {:?}",
            response.result
        );
        let out = response.result.unwrap();
        assert_eq!(
            out["count"],
            serde_json::json!(3),
            "responder saw all 3 chunks"
        );
        assert_eq!(
            out["last"],
            serde_json::json!({"chunk": 3}),
            "last chunk matches"
        );

        drop(client);
        let _ = server_handle.await;
    }

    #[tokio::test]
    async fn publish_end_to_end_unknown_op_returns_not_found() {
        use crate::core::auth::IdentityProvider;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::duplex_connection_pair;
        use crate::registry::registration::make_sink_handler;

        struct NoopIdProvider;
        impl IdentityProvider for NoopIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
                None
            }
            fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
                None
            }
        }

        let registry = Arc::new(OperationRegistry::new());

        let (client_conn, server_conn) = duplex_connection_pair(64 * 1024);
        let client = CallConnection::new(client_conn);

        let dp = Dispatcher::new(Arc::clone(&registry), Arc::new(NoopIdProvider));
        let server_call_conn = Arc::new(CallConnection::new(server_conn));
        let server_handle = tokio::spawn(async move {
            dp.run_loop(server_call_conn).await;
        });

        let stream: Pin<Box<dyn Stream<Item = serde_json::Value> + Send>> =
            Box::pin(futures::stream::iter(vec![
                serde_json::json!({"chunk": 1}),
                serde_json::json!({"chunk": 2}),
            ]));
        let _unused_sink = make_sink_handler(|_, ctx, _| async move {
            ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
        });
        let response = client
            .publish("no/such/op", serde_json::json!({}), stream)
            .await;

        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }

        drop(client);
        let _ = server_handle.await;
    }

    // --- single-stream call mode (ADR-036 amendment) ------------------------

    #[test]
    fn new_single_stream_is_single_stream() {
        use crate::core::types::BiStream;

        let (client_end, _server_end) = tokio::io::duplex(64);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, _reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        assert!(conn.is_single_stream());
        assert!(conn.single_stream_writer().is_some());
        assert!(conn.connection().is_some());
    }

    #[test]
    fn new_is_not_single_stream() {
        let conn = CallConnection::new(stub_connection());
        assert!(!conn.is_single_stream());
        assert!(conn.single_stream_writer().is_none());
    }

    #[tokio::test]
    async fn single_stream_call_round_trip() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let pending = Arc::clone(conn.pending());
        tokio::spawn(async move {
            read_single_stream_until_closed(reader, &pending).await;
        });

        let (server_recv, server_send) = tokio::io::split(server_end);
        tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let mut writer = FrameFramedWriter::new(server_send);
            let envelope = reader.read_frame().await.unwrap();
            assert_eq!(envelope.r#type, EVENT_REQUESTED);
            let response =
                EventEnvelope::responded(&envelope.id, serde_json::json!({"result": 42}));
            writer.write_frame(&response).await.unwrap();
        });

        let response = conn.call("test/echo", serde_json::json!({"x": 1})).await;
        assert!(response.result.is_ok());
        assert_eq!(response.result.unwrap(), serde_json::json!({"result": 42}));
    }

    #[tokio::test]
    async fn single_stream_call_receives_error() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let pending = Arc::clone(conn.pending());
        tokio::spawn(async move {
            read_single_stream_until_closed(reader, &pending).await;
        });

        let (server_recv, server_send) = tokio::io::split(server_end);
        tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let mut writer = FrameFramedWriter::new(server_send);
            let envelope = reader.read_frame().await.unwrap();
            let err = CallError::not_found("test/missing");
            writer
                .write_frame(&EventEnvelope::error(&envelope.id, &err))
                .await
                .unwrap();
        });

        let response = conn.call("test/missing", serde_json::json!({})).await;
        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn single_stream_subscribe_receives_multiple_events() {
        use crate::core::types::BiStream;
        use futures::stream::StreamExt;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let pending = Arc::clone(conn.pending());
        tokio::spawn(async move {
            read_single_stream_until_closed(reader, &pending).await;
        });

        let (server_recv, server_send) = tokio::io::split(server_end);
        tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let mut writer = FrameFramedWriter::new(server_send);
            let envelope = reader.read_frame().await.unwrap();
            let id = envelope.id.clone();
            writer
                .write_frame(&EventEnvelope::responded(&id, serde_json::json!("a")))
                .await
                .unwrap();
            writer
                .write_frame(&EventEnvelope::responded(&id, serde_json::json!("b")))
                .await
                .unwrap();
            writer
                .write_frame(&EventEnvelope::completed(&id))
                .await
                .unwrap();
        });

        let mut stream = conn.subscribe("test/events", serde_json::json!({})).await;
        let a = stream.next().await.unwrap();
        assert_eq!(a.result.unwrap(), serde_json::json!("a"));
        let b = stream.next().await.unwrap();
        assert_eq!(b.result.unwrap(), serde_json::json!("b"));
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn single_stream_subscribe_receives_error() {
        use crate::core::types::BiStream;
        use futures::stream::StreamExt;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let pending = Arc::clone(conn.pending());
        tokio::spawn(async move {
            read_single_stream_until_closed(reader, &pending).await;
        });

        let (server_recv, server_send) = tokio::io::split(server_end);
        tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let mut writer = FrameFramedWriter::new(server_send);
            let envelope = reader.read_frame().await.unwrap();
            let err = CallError::timeout("too slow");
            writer
                .write_frame(&EventEnvelope::error(&envelope.id, &err))
                .await
                .unwrap();
        });

        let mut stream = conn.subscribe("test/events", serde_json::json!({})).await;
        let item = stream.next().await.unwrap();
        match item.result {
            Err(e) => {
                assert_eq!(e.code, "TIMEOUT");
                assert!(e.retryable);
            }
            other => panic!("expected TIMEOUT, got {other:?}"),
        }
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn single_stream_publish_sends_chunks_and_receives_response() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let pending = Arc::clone(conn.pending());
        tokio::spawn(async move {
            read_single_stream_until_closed(reader, &pending).await;
        });

        let (server_recv, server_send) = tokio::io::split(server_end);
        let server_handle = tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let mut writer = FrameFramedWriter::new(server_send);

            let requested = reader.read_frame().await.unwrap();
            assert_eq!(requested.r#type, EVENT_REQUESTED);
            let id = requested.id.clone();

            let chunk1 = reader.read_frame().await.unwrap();
            assert_eq!(chunk1.r#type, EVENT_PUBLISHED);
            let chunk2 = reader.read_frame().await.unwrap();
            assert_eq!(chunk2.r#type, EVENT_PUBLISHED);

            let completed = reader.read_frame().await.unwrap();
            assert_eq!(completed.r#type, EVENT_COMPLETED);

            writer
                .write_frame(&EventEnvelope::responded(
                    &id,
                    serde_json::json!({"ok": true}),
                ))
                .await
                .unwrap();
        });

        let chunks = vec![serde_json::json!({"c": 1}), serde_json::json!({"c": 2})];
        let stream: Pin<Box<dyn Stream<Item = Value> + Send>> =
            Box::pin(futures::stream::iter(chunks));
        let response = conn
            .publish("test/upload", serde_json::json!({}), stream)
            .await;
        assert!(response.result.is_ok());
        assert_eq!(response.result.unwrap(), serde_json::json!({"ok": true}));

        server_handle.await.unwrap();
    }

    #[tokio::test]
    async fn single_stream_publish_receives_error() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let pending = Arc::clone(conn.pending());
        tokio::spawn(async move {
            read_single_stream_until_closed(reader, &pending).await;
        });

        let (server_recv, server_send) = tokio::io::split(server_end);
        tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let mut writer = FrameFramedWriter::new(server_send);
            let envelope = reader.read_frame().await.unwrap();
            let err = CallError::invalid_input("bad data");
            writer
                .write_frame(&EventEnvelope::error(&envelope.id, &err))
                .await
                .unwrap();
        });

        let chunks = vec![serde_json::json!({"c": 1})];
        let stream: Pin<Box<dyn Stream<Item = Value> + Send>> =
            Box::pin(futures::stream::iter(chunks));
        let response = conn
            .publish("test/upload", serde_json::json!({}), stream)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INVALID_INPUT"),
            other => panic!("expected INVALID_INPUT, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn single_stream_abort_sends_frame_and_removes_entry() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, _reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);

        let _rx = conn.pending().lock().register_call(
            "abort-me".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        assert!(conn.pending().lock().contains("abort-me"));

        let (server_recv, _server_send) = tokio::io::split(server_end);
        let server_handle = tokio::spawn(async move {
            let mut reader = FrameFramedReader::new(server_recv);
            let envelope = reader.read_frame().await.unwrap();
            assert_eq!(envelope.r#type, EVENT_ABORTED);
            assert_eq!(envelope.id, "abort-me");
        });

        conn.abort("abort-me").await;
        assert!(!conn.pending().lock().contains("abort-me"));

        server_handle.await.unwrap();
    }

    #[tokio::test]
    async fn shared_frame_writer_round_trip() {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let writer = Arc::new(SharedFrameWriter::new(Box::new(client_end)));

        let envelope = EventEnvelope::requested("req-1", serde_json::json!({"x": 1}));
        writer.write_frame(&envelope).await.unwrap();

        let mut reader = FrameFramedReader::new(server_end);
        let read = reader.read_frame().await.unwrap();
        assert_eq!(read, envelope);
    }

    // CF-001 — a transport write failure before the request frame could be
    // delivered maps to retryable CONNECTION_CLOSED (the call provably
    // never reached the producer). Mid-publish failures stay INTERNAL.

    /// AsyncWrite that always errors — simulates a dead mux mid-call.
    struct BrokenWrite;
    impl AsyncWrite for BrokenWrite {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            _buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            Poll::Ready(Err(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "mux dead",
            )))
        }
        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }
        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    fn broken_write_bi_stream() -> crate::core::types::BiStream {
        // The joined stream holds the read end; the write half always errors.
        let (read_end, _write_end) = tokio::io::duplex(1024);
        crate::core::types::BiStream::from_joined(read_end, BrokenWrite)
    }

    /// AsyncWrite whose writes succeed but whose Nth `poll_flush` fails.
    /// `FrameFramedWriter::write_frame` flushes exactly once per frame, so
    /// `remaining_ok_flushes = k` lets frames 1..=k land and fails inside
    /// frame k+1 — a deterministic mid-stream write failure.
    struct FailOnFlushN {
        remaining_ok_flushes: std::sync::atomic::AtomicUsize,
    }
    impl AsyncWrite for FailOnFlushN {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            Poll::Ready(Ok(buf.len()))
        }
        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            let prev = self
                .remaining_ok_flushes
                .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
            if prev == 0 {
                Poll::Ready(Err(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "mux dead",
                )))
            } else {
                Poll::Ready(Ok(()))
            }
        }
        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    #[tokio::test]
    async fn single_stream_call_write_failure_is_retryable_connection_closed() {
        let bi_stream = broken_write_bi_stream();
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);
        drop(reader);

        let response = conn.call("test/echo", serde_json::json!({})).await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "CONNECTION_CLOSED");
                assert!(e.retryable);
            }
            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn multi_stream_call_write_request_failure_is_retryable() {
        // A `Connection` whose open_bi succeeds but whose write half is
        // dead: the request write fails, the call provably undelivered.
        use crate::core::types::StreamError;
        use crate::core::types::{BiStream, BidiStreamSource, Connection};
        use std::net::SocketAddr;

        struct OneBiStream(std::sync::Mutex<Option<crate::core::types::BiStream>>);
        #[async_trait::async_trait]
        impl BidiStreamSource for OneBiStream {
            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
                Err(StreamError::ConnectionClosed)
            }
            async fn open_bi(&self) -> Result<BiStream, StreamError> {
                let stream = self.0.lock().expect("open once").take();
                stream.ok_or(StreamError::StreamClosed)
            }
            fn remote_addr(&self) -> Option<SocketAddr> {
                None
            }
            fn close(&self, _code: u32, _reason: &str) {}
        }

        let conn = Connection::from_source(
            OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))),
            b"alk/call".to_vec(),
        );
        let call_conn = CallConnection::new(conn);
        let response = call_conn.call("test/echo", serde_json::json!({})).await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "CONNECTION_CLOSED");
                assert!(e.retryable);
            }
            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn multi_stream_publish_request_frame_write_failure_is_retryable() {
        use crate::core::types::StreamError;
        use crate::core::types::{BiStream, BidiStreamSource, Connection};
        use std::net::SocketAddr;

        struct OneBiStream(std::sync::Mutex<Option<crate::core::types::BiStream>>);
        #[async_trait::async_trait]
        impl BidiStreamSource for OneBiStream {
            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
                Err(StreamError::ConnectionClosed)
            }
            async fn open_bi(&self) -> Result<BiStream, StreamError> {
                let stream = self.0.lock().expect("open once").take();
                stream.ok_or(StreamError::StreamClosed)
            }
            fn remote_addr(&self) -> Option<SocketAddr> {
                None
            }
            fn close(&self, _code: u32, _reason: &str) {}
        }

        let conn = Connection::from_source(
            OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))),
            b"alk/call".to_vec(),
        );
        let call_conn = CallConnection::new(conn);
        let response = call_conn
            .publish(
                "fs/upload",
                serde_json::json!({}),
                Box::pin(futures::stream::iter(vec![serde_json::json!({"c": 1})])),
            )
            .await;
        match response.result {
            Err(e) => {
                // The pump died on the *request* frame — nothing delivered.
                assert_eq!(e.code, "CONNECTION_CLOSED");
                assert!(e.retryable);
            }
            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
        }
    }

    #[test]
    fn connection_closed_error_is_retryable() {
        let err = CallError::connection_closed("test");
        assert_eq!(err.code, "CONNECTION_CLOSED");
        assert!(err.retryable);
    }

    #[tokio::test]
    async fn single_stream_subscribe_write_failure_is_retryable_connection_closed() {
        use futures::stream::StreamExt;

        let bi_stream = broken_write_bi_stream();
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);
        drop(reader);

        let mut stream = conn.subscribe("test/events", serde_json::json!({})).await;
        let item = stream.next().await.unwrap();
        match item.result {
            Err(e) => {
                assert_eq!(e.code, "CONNECTION_CLOSED");
                assert!(e.retryable);
            }
            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
        }
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn multi_stream_subscribe_write_failure_is_retryable_connection_closed() {
        use crate::core::types::StreamError;
        use crate::core::types::{BiStream, BidiStreamSource, Connection};
        use futures::stream::StreamExt;
        use std::net::SocketAddr;

        struct OneBiStream(std::sync::Mutex<Option<crate::core::types::BiStream>>);
        #[async_trait::async_trait]
        impl BidiStreamSource for OneBiStream {
            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
                Err(StreamError::ConnectionClosed)
            }
            async fn open_bi(&self) -> Result<BiStream, StreamError> {
                let stream = self.0.lock().expect("open once").take();
                stream.ok_or(StreamError::StreamClosed)
            }
            fn remote_addr(&self) -> Option<SocketAddr> {
                None
            }
            fn close(&self, _code: u32, _reason: &str) {}
        }

        let conn = Connection::from_source(
            OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))),
            b"alk/call".to_vec(),
        );
        let call_conn = CallConnection::new(conn);

        let mut stream = call_conn
            .subscribe("test/events", serde_json::json!({}))
            .await;
        let item = stream.next().await.unwrap();
        match item.result {
            Err(e) => {
                assert_eq!(e.code, "CONNECTION_CLOSED");
                assert!(e.retryable);
            }
            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
        }
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn single_stream_publish_request_frame_write_failure_is_retryable() {
        let bi_stream = broken_write_bi_stream();
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);
        drop(reader);

        let response = conn
            .publish(
                "fs/upload",
                serde_json::json!({}),
                Box::pin(futures::stream::iter(vec![serde_json::json!({"c": 1})])),
            )
            .await;
        match response.result {
            Err(e) => {
                // The request frame never landed — nothing delivered, retry safe.
                assert_eq!(e.code, "CONNECTION_CLOSED");
                assert!(e.retryable);
            }
            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn single_stream_publish_midpublish_write_failure_stays_internal() {
        // Flush succeeds for the request frame (1) + first chunk (2), then
        // fails inside the second chunk — delivery is ambiguous from there.
        let bi_stream = crate::core::types::BiStream::from_joined(
            tokio::io::empty(),
            FailOnFlushN {
                remaining_ok_flushes: std::sync::atomic::AtomicUsize::new(2),
            },
        );
        let (writer, reader) = split_single_stream(bi_stream);
        let conn = CallConnection::new_single_stream(stub_connection(), writer);
        drop(reader);

        let chunks = vec![serde_json::json!({"c": 1}), serde_json::json!({"c": 2})];
        let stream: Pin<Box<dyn Stream<Item = Value> + Send>> =
            Box::pin(futures::stream::iter(chunks));
        let response = conn
            .publish("fs/upload", serde_json::json!({}), stream)
            .await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "INTERNAL");
                assert!(!e.retryable, "mid-publish failure must not be retryable");
            }
            other => panic!("expected non-retryable INTERNAL, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn multi_stream_publish_midpublish_write_failure_stays_internal() {
        use crate::core::types::StreamError;
        use crate::core::types::{BiStream, BidiStreamSource, Connection};
        use std::net::SocketAddr;

        struct OneBiStream(std::sync::Mutex<Option<crate::core::types::BiStream>>);
        #[async_trait::async_trait]
        impl BidiStreamSource for OneBiStream {
            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
                Err(StreamError::ConnectionClosed)
            }
            async fn open_bi(&self) -> Result<BiStream, StreamError> {
                let stream = self.0.lock().expect("open once").take();
                stream.ok_or(StreamError::StreamClosed)
            }
            fn remote_addr(&self) -> Option<SocketAddr> {
                None
            }
            fn close(&self, _code: u32, _reason: &str) {}
        }

        let conn = Connection::from_source(
            OneBiStream(std::sync::Mutex::new(Some(
                crate::core::types::BiStream::from_joined(
                    tokio::io::empty(),
                    FailOnFlushN {
                        remaining_ok_flushes: std::sync::atomic::AtomicUsize::new(2),
                    },
                ),
            ))),
            b"alk/call".to_vec(),
        );
        let call_conn = CallConnection::new(conn);

        let chunks = vec![serde_json::json!({"c": 1}), serde_json::json!({"c": 2})];
        let stream: Pin<Box<dyn Stream<Item = Value> + Send>> =
            Box::pin(futures::stream::iter(chunks));
        let response = call_conn
            .publish("fs/upload", serde_json::json!({}), stream)
            .await;
        match response.result {
            Err(e) => {
                // The pump died on a *chunk* frame — the request landed,
                // delivery is ambiguous, INTERNAL (non-retryable) is correct.
                assert_eq!(e.code, "INTERNAL");
                assert!(!e.retryable);
            }
            other => panic!("expected non-retryable INTERNAL, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn shared_frame_writer_concurrent_writes_do_not_interleave() {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let writer = Arc::new(SharedFrameWriter::new(Box::new(client_end)));

        let w1 = Arc::clone(&writer);
        let w2 = Arc::clone(&writer);

        let h1 = tokio::spawn(async move {
            w1.write_frame(&EventEnvelope::requested("a", serde_json::json!({"n": 1})))
                .await
                .unwrap();
        });
        let h2 = tokio::spawn(async move {
            w2.write_frame(&EventEnvelope::requested("b", serde_json::json!({"n": 2})))
                .await
                .unwrap();
        });

        h1.await.unwrap();
        h2.await.unwrap();

        let mut reader = FrameFramedReader::new(server_end);
        let mut ids = Vec::new();
        for _ in 0..2 {
            let envelope = reader.read_frame().await.unwrap();
            ids.push(envelope.id);
        }
        assert!(ids.contains(&"a".to_string()));
        assert!(ids.contains(&"b".to_string()));
    }

    #[tokio::test]
    async fn split_single_stream_writer_to_server_round_trip() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (writer, _reader) = split_single_stream(bi_stream);

        let envelope = EventEnvelope::requested("req-1", serde_json::json!({"x": 1}));
        writer.write_frame(&envelope).await.unwrap();

        let mut server_reader = FrameFramedReader::new(server_end);
        let read = server_reader.read_frame().await.unwrap();
        assert_eq!(read, envelope);
    }

    #[tokio::test]
    async fn split_single_stream_server_to_reader_round_trip() {
        use crate::core::types::BiStream;

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let bi_stream = BiStream::from_bidi(client_end);
        let (_writer, reader) = split_single_stream(bi_stream);

        let envelope = EventEnvelope::responded("req-1", serde_json::json!({"x": 1}));
        let mut server_writer = FrameFramedWriter::new(server_end);
        server_writer.write_frame(&envelope).await.unwrap();
        drop(server_writer);

        let mut client_reader = FrameFramedReader::new(reader);
        let read = client_reader.read_frame().await.unwrap();
        assert_eq!(read, envelope);
    }

    #[tokio::test]
    async fn read_single_stream_until_closed_dispatches_frames() {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let pending = empty_pending();
        let rx = pending.lock().register_call(
            "req-ss-1".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );

        let mut server_writer = FrameFramedWriter::new(server_end);
        server_writer
            .write_frame(&EventEnvelope::responded(
                "req-ss-1",
                serde_json::json!({"v": 99}),
            ))
            .await
            .unwrap();
        drop(server_writer);

        let pending2 = Arc::clone(&pending);
        read_single_stream_until_closed(Box::new(client_end), &pending2).await;

        let result = tokio::time::timeout(Duration::from_millis(100), rx).await;
        match result {
            Ok(Ok(Ok(value))) => assert_eq!(value, serde_json::json!({"v": 99})),
            other => panic!("expected Ok({{v:99}}), got {other:?}"),
        }
    }

    // --- stream-per-request call mode (multi-stream transport) --------------

    #[tokio::test]
    async fn call_with_payload_stream_per_request_round_trip() {
        use crate::core::auth::IdentityProvider;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::duplex_connection_pair;
        struct NoopIdProvider;
        impl IdentityProvider for NoopIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
                None
            }
            fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
                None
            }
        }

        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("test/echo"),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let (client_conn, server_conn) = duplex_connection_pair(64 * 1024);
        let client = CallConnection::new(client_conn);

        let dp = Dispatcher::new(Arc::clone(&registry), Arc::new(NoopIdProvider));
        let server_call_conn = Arc::new(CallConnection::new(server_conn));
        let server_handle = tokio::spawn(async move {
            dp.run_loop(server_call_conn).await;
        });

        let response = client.call("test/echo", serde_json::json!({"x": 1})).await;
        assert!(response.result.is_ok());
        assert_eq!(response.result.unwrap(), serde_json::json!({"x": 1}));

        drop(client);
        let _ = server_handle.await;
    }

    #[tokio::test]
    async fn call_with_payload_stream_per_request_unknown_op_returns_not_found() {
        use crate::core::auth::IdentityProvider;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::duplex_connection_pair;

        struct NoopIdProvider;
        impl IdentityProvider for NoopIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
                None
            }
            fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
                None
            }
        }

        let registry = Arc::new(OperationRegistry::new());

        let (client_conn, server_conn) = duplex_connection_pair(64 * 1024);
        let client = CallConnection::new(client_conn);

        let dp = Dispatcher::new(Arc::clone(&registry), Arc::new(NoopIdProvider));
        let server_call_conn = Arc::new(CallConnection::new(server_conn));
        let server_handle = tokio::spawn(async move {
            dp.run_loop(server_call_conn).await;
        });

        let response = client.call("no/such/op", serde_json::json!({})).await;
        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }

        drop(client);
        let _ = server_handle.await;
    }

    #[tokio::test]
    async fn subscribe_with_payload_stream_per_request_receives_events() {
        use crate::core::auth::IdentityProvider;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::duplex_connection_pair;
        use crate::registry::registration::make_streaming_handler;
        use futures::stream::StreamExt;

        struct NoopIdProvider;
        impl IdentityProvider for NoopIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
                None
            }
            fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
                None
            }
        }

        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                OperationSpec::new(
                    "events/stream",
                    OperationType::Sub,
                    Visibility::External,
                    serde_json::json!({}),
                    serde_json::json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Stream(make_streaming_handler(|_input, ctx| {
                    futures::stream::iter(vec![
                        ResponseEnvelope::ok(ctx.request_id.clone(), serde_json::json!("a")),
                        ResponseEnvelope::ok(ctx.request_id, serde_json::json!("b")),
                    ])
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let registry = Arc::new(registry);

        let (client_conn, server_conn) = duplex_connection_pair(64 * 1024);
        let client = CallConnection::new(client_conn);

        let dp = Dispatcher::new(Arc::clone(&registry), Arc::new(NoopIdProvider));
        let server_call_conn = Arc::new(CallConnection::new(server_conn));
        let server_handle = tokio::spawn(async move {
            dp.run_loop(server_call_conn).await;
        });

        let mut stream = client
            .subscribe("events/stream", serde_json::json!({}))
            .await;
        let a = stream.next().await.unwrap();
        assert_eq!(a.result.unwrap(), serde_json::json!("a"));
        let b = stream.next().await.unwrap();
        assert_eq!(b.result.unwrap(), serde_json::json!("b"));
        assert!(stream.next().await.is_none());

        drop(client);
        let _ = server_handle.await;
    }

    #[tokio::test]
    async fn abort_stream_per_request_sends_frame_and_removes_entry() {
        use crate::core::auth::IdentityProvider;
        use crate::protocol::dispatch::Dispatcher;
        use crate::protocol::duplex_connection_pair;

        struct NoopIdProvider;
        impl IdentityProvider for NoopIdProvider {
            fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
                None
            }
            fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
                None
            }
        }

        let registry = Arc::new(OperationRegistry::new());

        let (client_conn, server_conn) = duplex_connection_pair(64 * 1024);
        let client = CallConnection::new(client_conn);

        let dp = Dispatcher::new(Arc::clone(&registry), Arc::new(NoopIdProvider));
        let server_call_conn = Arc::new(CallConnection::new(server_conn));
        let server_handle = tokio::spawn(async move {
            dp.run_loop(server_call_conn).await;
        });

        let _rx = client.pending().lock().register_call(
            "abort-spr".to_string(),
            Some(Instant::now() + Duration::from_secs(30)),
            None,
        );
        assert!(client.pending().lock().contains("abort-spr"));

        client.abort("abort-spr").await;
        assert!(!client.pending().lock().contains("abort-spr"));

        drop(client);
        let _ = server_handle.await;
    }
}