agent_engine/session/

socket_transport.rs

//! `SocketTransport` — the client side of the daemon UDS (PLAN-phase2 §2.10).
//!
//! Line-JSON frames (`wire.rs`). A writer task owns the write half; the
//! reader lives on the transport. Backpressure mirrors `LocalTransport`:
//! `send` fails with `Backpressure` when the bounded writer queue is full and
//! `Closed` once the socket is gone — never a silent drop.
//!
//! Tracing here logs frame *types* only (`ClientFrame`'s redacting `Debug`).

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::mpsc;

use super::handle::CMD_CHAN_CAP;
use super::transport::{ClientTransport, TransportError, ATTACH_TIMEOUT, ATTACH_TIMEOUT_PARKED};
use super::types::*;
use super::view::RuntimeView;
use super::wire::*;
use crate::ipc::{ClientReadHalf, ClientWriteHalf};

pub const CONNECT_TIMEOUT: Duration = Duration::from_millis(500);
pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(2);
/// How long `close` waits for the daemon's answer: its queue retry, the
/// actor's reply and the session's end (`daemon/close.rs` budgets), plus a
/// margin.
pub const CLOSE_TIMEOUT: Duration = Duration::from_secs(
    crate::daemon::close::CLOSE_SEND_RETRY.as_secs()
        + crate::daemon::close::CLOSE_REPLY_BUDGET.as_secs()
        + crate::daemon::close::CLOSE_END_BUDGET.as_secs()
        + 2,
);
/// How long `pin` waits for the daemon's answer: its longest wait (a parked
/// session unparking) plus a margin.
pub const PIN_TIMEOUT: Duration =
    Duration::from_secs(crate::daemon::pin::PIN_UNPARK_BUDGET.as_secs() + 2);
/// How long `rename` waits for the daemon's answer: its queue retry and the
/// actor's reply (`daemon/rename.rs` budgets), plus a margin.
pub const RENAME_TIMEOUT: Duration = Duration::from_secs(
    crate::daemon::rename::RENAME_SEND_RETRY.as_secs() + crate::daemon::rename::RENAME_REPLY_BUDGET.as_secs() + 2,
);
/// How long `reload_extension` waits: the daemon first waits for the
/// extension manager's lock, then stops the old process and runs the new
/// one's start-up handshake (each bounded by the extension runtime's own
/// timeouts, around 15 s together). Generous, because giving up early only
/// hides the answer: the daemon finishes the reload either way.
pub const RELOAD_EXTENSION_TIMEOUT: Duration = Duration::from_secs(60);

/// A connection past `Hello`/`Welcome`, before `Attach`.
pub struct Connected {
    reader: BufReader<ClientReadHalf>,
    writer: mpsc::Sender<ClientFrame>,
    writer_task: tokio::task::JoinHandle<()>,
    pub welcome: Welcome,
    /// Remembered for `reconnect`.
    path: PathBuf,
    hello: Hello,
    buf: FrameBuf,
}

pub struct Pong {
    pub pid: u32,
    pub uptime_s: u64,
    pub sessions: usize,
}

/// What `synaps auth state` got from the daemon.
#[derive(Debug)]
pub enum AuthStateReply {
    Report(agent_core::auth::AuthStateReport),
    /// The daemon predates `AuthState`: it answered the frame the way an
    /// old daemon answers any frame it cannot decode. Its version and build
    /// (empty when the daemon does not report one), from its `Welcome`.
    DaemonPredates { daemon_version: String, daemon_build: String },
}

/// An old daemon's reply to `AuthState` (daemon/conn.rs `Read::Bad`: serde's
/// unknown-variant error behind "malformed frame: "), and nothing else.
pub fn predates_auth_state(message: &str) -> bool {
    message.starts_with("malformed frame: unknown variant `auth_state`")
}

/// Reused line buffer for `read_frame`: a big `Attached`/`MessageHistory`
/// frame grows it once; after any frame > 1 MiB it shrinks back to 64 KiB so
/// the peak is not retained for the life of the connection (§2.1 #1).
#[derive(Default)]
pub struct FrameBuf {
    line: Vec<u8>,
    /// Length of the last decoded frame (bytes, without the newline).
    last_frame_bytes: usize,
}

const FRAME_BUF_KEEP: usize = 64 * 1024;
const FRAME_BUF_SHRINK_ABOVE: usize = 1024 * 1024;

impl FrameBuf {
    fn after_frame(&mut self, n: usize) {
        self.last_frame_bytes = n;
        if self.line.capacity() > FRAME_BUF_SHRINK_ABOVE {
            self.line = Vec::new();
            self.line.reserve(FRAME_BUF_KEEP);
        }
    }
}

async fn read_frame(reader: &mut BufReader<ClientReadHalf>, buf: &mut FrameBuf) -> Result<Option<DaemonFrame>, TransportError> {
    // Cancel-safe: the TUI polls this inside `select!`, so a partial line
    // stays in `buf.line` across a dropped future and is cleared only once
    // a whole frame has been taken out of it.
    loop {
        // `take` bounds the read so an oversize frame cannot grow memory.
        let limit = (MAX_FRAME_BYTES as u64 + 1).saturating_sub(buf.line.len() as u64);
        let n = tokio::io::AsyncReadExt::take(&mut *reader, limit)
            .read_until(b'\n', &mut buf.line)
            .await?;
        if n == 0 {
            return Ok(None);
        }
        if buf.line.len() > MAX_FRAME_BYTES {
            buf.line.clear();
            return Err(TransportError::Protocol(frame_limit_msg()));
        }
        let decoded = {
            let t = std::str::from_utf8(&buf.line)
                .map_err(|e| TransportError::Protocol(format!("frame is not utf-8: {e}")))
                .map(str::trim_end);
            match t {
                Ok("") => None,
                Ok(t) => Some((decode_line::<DaemonFrame>(t).map(Some).map_err(TransportError::Protocol), t.len())),
                Err(e) => Some((Err(e), 0)),
            }
        };
        buf.line.clear();
        let Some((decoded, len)) = decoded else { continue };
        buf.after_frame(len);
        return decoded;
    }
}

fn spawn_writer(mut w: ClientWriteHalf) -> (mpsc::Sender<ClientFrame>, tokio::task::JoinHandle<()>) {
    let (tx, mut rx) = mpsc::channel::<ClientFrame>(CMD_CHAN_CAP);
    let task = tokio::spawn(async move {
        while let Some(frame) = rx.recv().await {
            let line = match encode_line(&frame) {
                Ok(l) => l,
                Err(e) => {
                    tracing::warn!(frame = ?frame, "socket_transport: encode failed: {e}");
                    continue;
                }
            };
            if let Err(e) = w.write_all(line.as_bytes()).await {
                tracing::debug!("socket_transport: write failed: {e}");
                break;
            }
            if matches!(frame, ClientFrame::Bye) {
                let _ = w.shutdown().await;
                break;
            }
        }
    });
    (tx, task)
}

impl Connected {
    /// Hello/Welcome only. `Refused{Version}` → `TransportError::Version`;
    /// any other refusal → `Protocol`.
    pub async fn connect(path: &Path, hello: Hello) -> Result<Self, TransportError> {
        let stream = tokio::time::timeout(CONNECT_TIMEOUT, crate::ipc::connect(path))
            .await
            .map_err(|_| TransportError::Protocol("connect timed out".into()))??;
        let (r, w) = crate::ipc::split_client(stream);
        let mut reader = BufReader::new(r);
        let (writer, writer_task) = spawn_writer(w);
        let client_version = hello.protocol_version;
        let hello_copy = hello.clone();
        let mut buf = FrameBuf::default();
        writer
            .send(ClientFrame::Hello(hello))
            .await
            .map_err(|_| TransportError::Closed)?;
        let frame = tokio::time::timeout(HANDSHAKE_TIMEOUT, read_frame(&mut reader, &mut buf))
            .await
            .map_err(|_| TransportError::Protocol("handshake timed out".into()))??;
        match frame {
            Some(DaemonFrame::Welcome(welcome)) => {
                if welcome.protocol_version != client_version {
                    return Err(TransportError::Version { client: client_version, daemon: welcome.protocol_version });
                }
                Ok(Self { reader, writer, writer_task, welcome, path: path.to_path_buf(), hello: hello_copy, buf })
            }
            Some(DaemonFrame::Refused { reason: RefuseReason::Version { daemon_version, .. }, .. }) => {
                Err(TransportError::Version { client: client_version, daemon: daemon_version })
            }
            Some(DaemonFrame::Refused { reason, message }) => {
                Err(TransportError::Protocol(format!("refused ({reason:?}): {message}")))
            }
            Some(other) => Err(TransportError::Protocol(format!("expected welcome, got {other:?}"))),
            None => Err(TransportError::Closed),
        }
    }

    async fn request(&mut self, frame: ClientFrame) -> Result<DaemonFrame, TransportError> {
        self.writer.send(frame).await.map_err(|_| TransportError::Closed)?;
        tokio::time::timeout(HANDSHAKE_TIMEOUT, read_frame(&mut self.reader, &mut self.buf))
            .await
            .map_err(|_| TransportError::Protocol("reply timed out".into()))??
            .ok_or(TransportError::Closed)
    }

    pub async fn ping(&mut self) -> Result<Pong, TransportError> {
        match self.request(ClientFrame::Ping).await? {
            DaemonFrame::Pong { pid, uptime_s, sessions } => Ok(Pong { pid, uptime_s, sessions }),
            other => Err(TransportError::Protocol(format!("expected pong, got {other:?}"))),
        }
    }

    pub async fn sessions(&mut self) -> Result<Vec<SessionMeta>, TransportError> {
        match self.request(ClientFrame::Sessions).await? {
            DaemonFrame::SessionList { sessions } => Ok(sessions),
            other => Err(TransportError::Protocol(format!("expected session list, got {other:?}"))),
        }
    }

    /// `Purge` → the daemon runs `memstat::purge_arenas()`; replies `Pong`.
    pub async fn purge(&mut self) -> Result<Pong, TransportError> {
        match self.request(ClientFrame::Purge).await? {
            DaemonFrame::Pong { pid, uptime_s, sessions } => Ok(Pong { pid, uptime_s, sessions }),
            DaemonFrame::Error { message, .. } => Err(TransportError::Protocol(message)),
            other => Err(TransportError::Protocol(format!("expected pong, got {other:?}"))),
        }
    }

    /// `AuthState` → the daemon broker's in-memory account state. A daemon
    /// that predates the frame is [`AuthStateReply::DaemonPredates`]; any
    /// other error is returned as itself.
    pub async fn auth_state(&mut self, query: agent_core::auth::AuthStateQuery) -> Result<AuthStateReply, TransportError> {
        match self.request(ClientFrame::AuthState(query)).await? {
            DaemonFrame::AuthState { report } => Ok(AuthStateReply::Report(report)),
            DaemonFrame::Error { session_id: None, message } if predates_auth_state(&message) => {
                Ok(AuthStateReply::DaemonPredates {
                    daemon_version: self.welcome.daemon_version.clone(),
                    daemon_build: self.welcome.daemon_build.clone(),
                })
            }
            DaemonFrame::Error { message, .. } => Err(TransportError::Protocol(message)),
            other => Err(TransportError::Protocol(format!("expected auth state, got {other:?}"))),
        }
    }

    pub async fn shutdown(&mut self, force: bool) -> Result<(), TransportError> {
        match self.request(ClientFrame::Shutdown { force }).await? {
            DaemonFrame::Bye { .. } => Ok(()),
            DaemonFrame::Error { message, .. } => Err(TransportError::Protocol(message)),
            other => Err(TransportError::Protocol(format!("expected bye, got {other:?}"))),
        }
    }

    /// `Close{id, force, attached}`: end one session with host authority.
    /// `Ok(text)` = the session ended (the actor's `/close` reply, with how
    /// to continue it). A refusal (work in flight, clients attached without
    /// `attached`, daemon reloading) → `Refused(text)`;
    /// no such session → `UnknownSession`; a daemon that predates `Close`
    /// → `DaemonTooOld`. Waits up to [`CLOSE_TIMEOUT`]: the session saves
    /// and runs its end hooks before the daemon answers.
    pub async fn close(&mut self, id: &SessionId, force: bool, attached: bool) -> Result<String, TransportError> {
        self.writer
            .send(ClientFrame::Close { session_id: id.clone(), force, attached })
            .await
            .map_err(|_| TransportError::Closed)?;
        let frame = tokio::time::timeout(CLOSE_TIMEOUT, read_frame(&mut self.reader, &mut self.buf))
            .await
            .map_err(|_| TransportError::Protocol("close reply timed out".into()))??;
        close_reply(frame, &self.welcome.daemon_version)
    }

    /// `Pin{id, on}`: pin or unpin one session with host authority.
    /// `Ok(on)` = the session's pin is now `on`. A refusal (daemon
    /// reloading, session live in another process) → `Refused(text)`; no
    /// such session → `UnknownSession`; a daemon that predates `Pin` →
    /// `DaemonTooOld`.
    pub async fn pin(&mut self, id: &SessionId, on: bool) -> Result<bool, TransportError> {
        self.writer
            .send(ClientFrame::Pin { session_id: id.clone(), on })
            .await
            .map_err(|_| TransportError::Closed)?;
        let frame = tokio::time::timeout(PIN_TIMEOUT, read_frame(&mut self.reader, &mut self.buf))
            .await
            .map_err(|_| TransportError::Protocol("pin reply timed out".into()))??;
        pin_reply(frame, &self.welcome.daemon_version)
    }

    /// `Rename{id, name, expect}`: name (`Some`) or unname (`None`) one
    /// session with host authority, without attaching. `expect` is the
    /// compare-and-set precondition (`Some(None)` = only if unnamed).
    /// `Ok(Ok(renamed))` = done; `Ok(Err(failure))` = the daemon says why
    /// not, or `Unknown` when no answer came (timeout, connection lost after
    /// sending: it may still apply). No such session → `UnknownSession`; a
    /// daemon that predates `Rename` → `DaemonTooOld` (never a hang: it
    /// answers the unknown frame at once).
    pub async fn rename(
        &mut self,
        id: &SessionId,
        name: Option<&str>,
        expect: Option<Option<String>>,
    ) -> Result<Result<Renamed, RenameFailure>, TransportError> {
        self.writer
            .send(ClientFrame::Rename { session_id: id.clone(), name: name.map(str::to_string), expect })
            .await
            .map_err(|_| TransportError::Closed)?;
        let unknown = |why: &str| {
            Ok(Err(RenameFailure {
                reason: crate::session::wire::RenameRefusal::Unknown,
                message: format!("{why}; the rename may still apply. Re-list (`synaps daemon sessions`) before retrying"),
                taken_by: None,
                current_name: None,
            }))
        };
        match tokio::time::timeout(RENAME_TIMEOUT, read_frame(&mut self.reader, &mut self.buf)).await {
            Err(_) => unknown("no reply from the daemon in time"),
            Ok(Ok(None)) | Ok(Err(TransportError::Closed)) => unknown("the daemon closed the connection before answering"),
            Ok(frame) => rename_reply(frame?, &self.welcome.daemon_version),
        }
    }

    /// `ReloadExtension{extension}`: stop that extension and start it again
    /// from what is installed now. `Ok(Ok(report))` = reloaded;
    /// `Ok(Err(failure))` = the daemon tried and says why not; a refusal
    /// (daemon reloading) → `Refused(text)`; a daemon that predates the
    /// frame → `DaemonTooOld`.
    pub async fn reload_extension(
        &mut self,
        extension: &str,
    ) -> Result<Result<ExtensionReloadReport, ExtensionReloadFailure>, TransportError> {
        self.writer
            .send(ClientFrame::ReloadExtension { extension: extension.to_string() })
            .await
            .map_err(|_| TransportError::Closed)?;
        let frame = tokio::time::timeout(RELOAD_EXTENSION_TIMEOUT, read_frame(&mut self.reader, &mut self.buf))
            .await
            .map_err(|_| TransportError::Protocol("extension reload reply timed out".into()))??;
        reload_extension_reply(frame, &self.welcome.daemon_version)
    }

    /// Say goodbye and drop the connection.
    pub async fn bye(self) {
        let _ = self.writer.send(ClientFrame::Bye).await;
        let _ = tokio::time::timeout(Duration::from_millis(200), self.writer_task).await;
    }
}

/// The daemon's answer to `Close`, as [`Connected::close`] reports it.
fn close_reply(frame: Option<DaemonFrame>, daemon_version: &str) -> Result<String, TransportError> {
    match frame {
        Some(DaemonFrame::Closed { text, .. }) => Ok(text),
        Some(DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message }) => {
            Err(TransportError::UnknownSession(message))
        }
        Some(DaemonFrame::Refused { message, .. }) => Err(TransportError::Refused(message)),
        // An older daemon cannot decode the frame; it answers with serde's
        // "unknown variant `close`" and keeps the connection open.
        Some(DaemonFrame::Error { message, .. }) if message.contains("unknown variant `close`") => {
            Err(TransportError::DaemonTooOld { daemon_version: daemon_version.to_string(), request: "daemon close" })
        }
        Some(DaemonFrame::Error { message, .. }) => Err(TransportError::Protocol(message)),
        Some(other) => Err(TransportError::Protocol(format!("expected closed, got {other:?}"))),
        None => Err(TransportError::Closed),
    }
}

/// The daemon's answer to `Pin`, as [`Connected::pin`] reports it.
fn pin_reply(frame: Option<DaemonFrame>, daemon_version: &str) -> Result<bool, TransportError> {
    match frame {
        Some(DaemonFrame::Pinned { on, .. }) => Ok(on),
        Some(DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message }) => {
            Err(TransportError::UnknownSession(message))
        }
        Some(DaemonFrame::Refused { message, .. }) => Err(TransportError::Refused(message)),
        // An older daemon cannot decode the frame; it answers with serde's
        // "unknown variant `pin`" and keeps the connection open.
        Some(DaemonFrame::Error { message, .. }) if message.contains("unknown variant `pin`") => {
            Err(TransportError::DaemonTooOld { daemon_version: daemon_version.to_string(), request: "daemon pin" })
        }
        Some(DaemonFrame::Error { message, .. }) => Err(TransportError::Protocol(message)),
        Some(other) => Err(TransportError::Protocol(format!("expected pinned, got {other:?}"))),
        None => Err(TransportError::Closed),
    }
}

/// A rename the daemon applied (`DaemonFrame::Renamed`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Renamed {
    /// The session's name now (`None` = cleared).
    pub name: Option<String>,
    /// The session is not persisted: nothing was written.
    pub in_memory_only: bool,
}

/// Why the daemon did not rename a session (`DaemonFrame::RenameFailed`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RenameFailure {
    pub reason: crate::session::wire::RenameRefusal,
    pub message: String,
    /// Journal id of the saved session that already has the name (`Taken`).
    pub taken_by: Option<String>,
    /// The session's name when the `expect` precondition failed (`Named`).
    pub current_name: Option<String>,
}

/// The daemon's answer to `Rename`, as [`Connected::rename`] reports it.
fn rename_reply(
    frame: Option<DaemonFrame>,
    daemon_version: &str,
) -> Result<Result<Renamed, RenameFailure>, TransportError> {
    match frame {
        Some(DaemonFrame::Renamed { name, in_memory_only, .. }) => Ok(Ok(Renamed { name, in_memory_only })),
        Some(DaemonFrame::RenameFailed { reason, message, taken_by, current_name, .. }) => {
            Ok(Err(RenameFailure { reason, message, taken_by, current_name }))
        }
        Some(DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message }) => {
            Err(TransportError::UnknownSession(message))
        }
        Some(DaemonFrame::Refused { message, .. }) => Err(TransportError::Refused(message)),
        // An older daemon cannot decode the frame; it answers with serde's
        // "unknown variant `rename`" and keeps the connection open.
        Some(DaemonFrame::Error { message, .. }) if message.contains("unknown variant `rename`") => {
            Err(TransportError::DaemonTooOld { daemon_version: daemon_version.to_string(), request: "daemon rename" })
        }
        Some(DaemonFrame::Error { message, .. }) => Err(TransportError::Protocol(message)),
        Some(other) => Err(TransportError::Protocol(format!("expected renamed, got {other:?}"))),
        None => Err(TransportError::Closed),
    }
}

pub use crate::extensions::manager::ExtensionReloadReport;

/// The daemon's reason an extension was not reloaded
/// (`DaemonFrame::ExtensionReloadFailed`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExtensionReloadFailure {
    pub message: String,
    /// The old instance was stopped before the new one failed to start: the
    /// extension is not loaded now. `false` = the running one (if any) was
    /// not touched.
    pub unloaded: bool,
}

/// The daemon's answer to `ReloadExtension`, as
/// [`Connected::reload_extension`] reports it.
fn reload_extension_reply(
    frame: Option<DaemonFrame>,
    daemon_version: &str,
) -> Result<Result<ExtensionReloadReport, ExtensionReloadFailure>, TransportError> {
    match frame {
        Some(DaemonFrame::ExtensionReloaded { report }) => Ok(Ok(report)),
        Some(DaemonFrame::ExtensionReloadFailed { message, unloaded, .. }) => {
            Ok(Err(ExtensionReloadFailure { message, unloaded }))
        }
        Some(DaemonFrame::Refused { message, .. }) => Err(TransportError::Refused(message)),
        // An older daemon cannot decode the frame; it answers with serde's
        // "unknown variant `reload_extension`" and keeps the connection open.
        Some(DaemonFrame::Error { message, .. }) if message.contains("unknown variant `reload_extension`") => {
            Err(TransportError::DaemonTooOld {
                daemon_version: daemon_version.to_string(),
                request: "daemon reload-extension",
            })
        }
        Some(DaemonFrame::Error { message, .. }) => Err(TransportError::Protocol(message)),
        Some(other) => Err(TransportError::Protocol(format!("expected extension_reloaded, got {other:?}"))),
        None => Err(TransportError::Closed),
    }
}

pub struct SocketTransport {
    reader: BufReader<ClientReadHalf>,
    writer: mpsc::Sender<ClientFrame>,
    _writer_task: tokio::task::JoinHandle<()>,
    session: SessionId,
    meta: SessionMeta,
    client: ClientId,
    view: Arc<arc_swap::ArcSwap<RuntimeView>>,
    pub welcome: Welcome,
    mode: AttachMode,
    input_owner: Option<ClientId>,
    roster: Roster,
    client_input: ClientInput,
    ended: bool,
    path: PathBuf,
    hello: Hello,
    /// `Reloading` seen; the next EOF is a reload, not a death.
    reload_pending: Option<u64>,
    /// Why `next_event` returned `None` (`Reloading{generation}` after a
    /// reload announcement) — the TUI branches to `reconnect` on it.
    last_error: Option<TransportError>,
    /// Local `api_messages` mirror (`HistoryMode::Full` only): seeded by
    /// `Attached`, updated by `Stream(MessageHistory)` and
    /// `QueryResult{DIGEST_RESYNC_QUERY_ID}`. `Conversation` arrives as a
    /// digest; the mirror fills it in. Empty forever in `Digest` mode.
    messages: Vec<crate::SharedMessage>,
    /// Digest awaiting a `Messages` re-query (hash miss; Full mode only).
    pending_digest: Option<ConversationDigest>,
    /// What we asked for in `Hello` (phase 4 §2.3).
    history: HistoryMode,
    buf: FrameBuf,
}

impl SocketTransport {
    /// Hello/Welcome only (50 ms connect + 2 s handshake budgets).
    pub async fn connect(path: &Path, hello: Hello) -> Result<Connected, TransportError> {
        Connected::connect(path, hello).await
    }

    /// Attach (existing or create) on a handshaken connection. `AttachRefused`
    /// and `DaemonFrame::Error` alike → `TransportError::Refused`. Waits
    /// `ATTACH_TIMEOUT_PARKED` when the daemon lists the session as `Parked`.
    pub async fn attach(mut conn: Connected, attach: Attach) -> Result<(Self, AttachSnapshot), TransportError> {
        let (mode, parked) = match &attach {
            Attach::Existing { session_id, mode } => (
                *mode,
                conn.welcome
                    .sessions
                    .iter()
                    .any(|m| &m.id == session_id && m.lifecycle == SessionLifecycle::Parked),
            ),
            Attach::Create { mode, .. } => (*mode, false),
        };
        conn.writer
            .send(ClientFrame::Attach(attach))
            .await
            .map_err(|_| TransportError::Closed)?;
        let budget = if parked { ATTACH_TIMEOUT_PARKED } else { ATTACH_TIMEOUT };
        let deadline = tokio::time::Instant::now() + budget;
        let attached = loop {
            let frame = tokio::time::timeout_at(deadline, read_frame(&mut conn.reader, &mut conn.buf))
                .await
                .map_err(|_| TransportError::Protocol("attach timed out".into()))??;
            match frame {
                Some(DaemonFrame::Attached(a)) => break a,
                Some(DaemonFrame::Error { message, .. }) => return Err(TransportError::Refused(message)),
                Some(DaemonFrame::Event(w)) => match w.event {
                    WireSessionEvent::AttachRefused { message } => return Err(TransportError::Refused(message)),
                    // Notices/lifecycle chatter may precede `Attached` (unpark).
                    _ => continue,
                },
                Some(DaemonFrame::Bye { .. }) | None => return Err(TransportError::Closed),
                Some(other) => return Err(TransportError::Protocol(format!("expected attached, got {other:?}"))),
            }
        };
        let (client, snapshot) = attached.into_snapshot();
        let meta = snapshot.meta.clone();
        let view = Arc::new(arc_swap::ArcSwap::from_pointee(snapshot.view.clone()));
        let history = conn.hello.client.history;
        let messages = match history {
            HistoryMode::Full => snapshot.conversation.api_messages.clone(),
            HistoryMode::Digest => Vec::new(),
        };
        Ok((
            Self {
                reader: conn.reader,
                writer: conn.writer,
                _writer_task: conn.writer_task,
                session: meta.id.clone(),
                meta,
                client,
                view,
                welcome: conn.welcome,
                mode,
                input_owner: snapshot.input_owner,
                roster: Roster::from_snapshot(&snapshot),
                client_input: snapshot.client_input,
                ended: false,
                path: conn.path,
                hello: conn.hello,
                reload_pending: None,
                last_error: None,
                messages,
                pending_digest: None,
                history,
                buf: conn.buf,
            },
            snapshot,
        ))
    }

    /// Why the last `next_event` returned `None` (if it was not a plain end).
    pub fn last_error(&self) -> Option<&TransportError> {
        self.last_error.as_ref()
    }

    /// `true` if the daemon announced a reload before EOF (graceful); `false`
    /// means a crash/SIGKILL (unexpected).
    pub fn is_reload_pending(&self) -> bool {
        self.reload_pending.is_some()
    }

    /// The pid the daemon reported in `Welcome`.
    pub fn daemon_pid(&self) -> u32 {
        self.welcome.pid
    }

    /// The reconnect budget from `SYNAPS_TUI_ATTACH_RECONNECT_SECS` (default 60).
    pub fn reconnect_budget() -> Duration {
        std::env::var("SYNAPS_TUI_ATTACH_RECONNECT_SECS")
            .ok()
            .and_then(|s| s.parse().ok())
            .map(Duration::from_secs)
            .unwrap_or(Duration::from_secs(60))
    }

    /// One attempt at reconnecting (Hello + Attach). On success replaces
    /// `self` and returns the snapshot. On `Refused` / `Version` the error is
    /// terminal (caller should stop retrying). Other errors are transient.
    pub async fn reconnect_once(&mut self, mode: AttachMode) -> Result<AttachSnapshot, TransportError> {
        let was_owner = self.input_owner == Some(self.client);
        let generation = self.reload_pending.unwrap_or(self.welcome.generation);
        let mut hello = self.hello.clone();
        hello.reconnect_of = Some(ClientReconnect {
            previous_client: self.client,
            session_id: self.session.clone(),
            was_owner,
            generation,
        });
        let mode = if was_owner { AttachMode::Takeover } else { mode };
        match Connected::connect(&self.path, hello).await {
            Ok(conn) => {
                match Self::attach(conn, Attach::Existing { session_id: self.session.clone(), mode }).await {
                    Ok((t, snap)) => {
                        *self = t;
                        Ok(snap)
                    }
                    Err(e) => Err(e),
                }
            }
            Err(e) => Err(e),
        }
    }

    // ── control fast path helpers (fresh connection each) ──

    pub async fn ping(path: &Path) -> Result<Pong, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let p = c.ping().await;
        c.bye().await;
        p
    }

    pub async fn sessions(path: &Path) -> Result<Vec<SessionMeta>, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let s = c.sessions().await;
        c.bye().await;
        s
    }

    pub async fn shutdown(path: &Path, force: bool) -> Result<(), TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        c.shutdown(force).await
    }

    pub async fn purge(path: &Path) -> Result<Pong, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let p = c.purge().await;
        c.bye().await;
        p
    }

    /// `Close{id, force, attached}` on a fresh control connection (see
    /// [`Connected::close`]).
    pub async fn close(path: &Path, id: &SessionId, force: bool, attached: bool) -> Result<String, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let r = c.close(id, force, attached).await;
        c.bye().await;
        r
    }

    /// `Pin{id, on}` on a fresh control connection (see [`Connected::pin`]).
    pub async fn pin(path: &Path, id: &SessionId, on: bool) -> Result<bool, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let r = c.pin(id, on).await;
        c.bye().await;
        r
    }

    /// `Rename{id, name}` on a fresh control connection (see
    /// [`Connected::rename`]).
    pub async fn rename(
        path: &Path,
        id: &SessionId,
        name: Option<&str>,
        expect: Option<Option<String>>,
    ) -> Result<Result<Renamed, RenameFailure>, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let r = c.rename(id, name, expect).await;
        c.bye().await;
        r
    }

    /// `AuthState` on a control connection (see [`Connected::auth_state`]).
    pub async fn auth_state(path: &Path, query: agent_core::auth::AuthStateQuery) -> Result<AuthStateReply, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        let reply = c.auth_state(query).await;
        c.bye().await;
        reply
    }

    /// `Reload{..}` on a control connection. `Ok(generation)` once the daemon
    /// answered `Bye{Reloading{generation}}` (it is about to exec);
    /// `Refused{ReloadRefused}` / `Error` → `Err`.
    pub async fn reload(path: &Path, now: bool, drain_secs: Option<u64>, exe: Option<PathBuf>) -> Result<u64, TransportError> {
        let mut c = Connected::connect(path, Hello::new(ClientKind::Attach)).await?;
        c.writer
            .send(ClientFrame::Reload { now, drain_secs, exe })
            .await
            .map_err(|_| TransportError::Closed)?;
        // Drain + checkpoint can take a while: wait up to drain + 60 s.
        let budget = Duration::from_secs(drain_secs.unwrap_or(30) + 60);
        let frame = tokio::time::timeout(budget, read_frame(&mut c.reader, &mut c.buf))
            .await
            .map_err(|_| TransportError::Protocol("reload reply timed out".into()))??;
        match frame {
            Some(DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation, .. }) }) => Ok(generation),
            Some(DaemonFrame::Bye { .. }) => Err(TransportError::Protocol("daemon said bye without a reload reason".into())),
            Some(DaemonFrame::Refused { reason: RefuseReason::ReloadRefused { why }, .. }) => Err(TransportError::Refused(why)),
            Some(DaemonFrame::Refused { message, .. }) | Some(DaemonFrame::Error { message, .. }) => Err(TransportError::Refused(message)),
            Some(other) => Err(TransportError::Protocol(format!("expected bye, got {other:?}"))),
            None => Err(TransportError::Closed),
        }
    }

    /// Detach cleanly: `Detach` + `Bye`. The turn keeps running in the daemon.
    pub async fn detach(&self) {
        let _ = self
            .writer
            .send(ClientFrame::Cmd { session_id: self.session.clone(), cmd: SessionCommand::Detach { client: self.client } })
            .await;
        let _ = self.writer.send(ClientFrame::Bye).await;
    }

    fn notice(&self, text: String) -> Envelope {
        Envelope { session_id: self.session.clone(), seq: u64::MAX, ts: chrono::Utc::now(), event: SessionEventWire::SystemNotice(text) }
    }

    /// The client's view of `api_messages` (kept current by the digest
    /// protocol in `Full` mode; always empty in `Digest` mode).
    pub fn messages(&self) -> &[crate::SharedMessage] {
        &self.messages
    }

    /// Length in bytes of the last frame decoded on this connection
    /// (the `Attached` frame right after `attach`).
    pub fn last_frame_bytes(&self) -> usize {
        self.buf.last_frame_bytes
    }

    pub fn history_mode(&self) -> HistoryMode {
        self.history
    }

    /// Turn a wire envelope into the in-process one, keeping the message
    /// mirror honest. `None` = swallowed (reserved query id, or a digest that
    /// needs a re-query first — the `Conversation` is re-emitted once the
    /// `Messages` result lands).
    fn absorb(&mut self, w: WireEnvelope) -> Option<Envelope> {
        if self.history == HistoryMode::Digest {
            return self.absorb_digest(w);
        }
        match w.event {
            WireSessionEvent::Conversation { digest } => {
                if digest.matches(&self.messages) {
                    let event = SessionEventWire::Conversation(digest.into_snapshot(self.messages.clone()));
                    return Some(Envelope { session_id: w.session_id, seq: w.seq, ts: w.ts, event });
                }
                if digest.messages_len == 0 {
                    self.messages.clear();
                    let event = SessionEventWire::Conversation(digest.into_snapshot(Vec::new()));
                    return Some(Envelope { session_id: w.session_id, seq: w.seq, ts: w.ts, event });
                }
                // Mirror drifted (compaction, abort repair, flushed events): re-fetch.
                let first = self.pending_digest.is_none();
                self.pending_digest = Some(digest);
                if first {
                    let _ = self.writer.try_send(ClientFrame::Cmd {
                        session_id: self.session.clone(),
                        cmd: SessionCommand::Query { id: DIGEST_RESYNC_QUERY_ID, query: SessionQuery::Messages },
                    });
                }
                None
            }
            WireSessionEvent::QueryResult { id, value } if id == DIGEST_RESYNC_QUERY_ID => {
                if let Ok(msgs) = serde_json::from_value::<Vec<serde_json::Value>>(value) {
                    self.messages = msgs.into_iter().map(Arc::new).collect();
                }
                let digest = self.pending_digest.take()?;
                if !digest.matches(&self.messages) {
                    tracing::debug!("socket_transport: messages re-query still disagrees with digest; using fetched history");
                }
                let event = SessionEventWire::Conversation(digest.into_snapshot(self.messages.clone()));
                Some(Envelope { session_id: w.session_id, seq: w.seq, ts: w.ts, event })
            }
            WireSessionEvent::QueryResult { id, .. } if id >= RESERVED_QUERY_ID_BASE => None,
            event => {
                let env = Envelope { session_id: w.session_id, seq: w.seq, ts: w.ts, event: event.into() };
                match &env.event {
                    SessionEventWire::Stream(crate::StreamEvent::Session(crate::SessionEvent::MessageHistory(m))) => {
                        self.messages = m.clone();
                    }
                    SessionEventWire::Attached { snapshot, .. } => {
                        self.messages = snapshot.conversation.api_messages.clone();
                    }
                    _ => {}
                }
                Some(env)
            }
        }
    }
}

impl SocketTransport {
    /// `HistoryMode::Digest`: no mirror, no `matches()` re-serialisation, no
    /// `Query{Messages}`. `Conversation` is passed through with an empty
    /// `api_messages` (`messages_len` from the digest); a stray
    /// `MessageHistory` is dropped (the daemon should not send one).
    fn absorb_digest(&mut self, w: WireEnvelope) -> Option<Envelope> {
        match w.event {
            WireSessionEvent::Conversation { digest } => {
                let event = SessionEventWire::Conversation(digest.into_snapshot(Vec::new()));
                Some(Envelope { session_id: w.session_id, seq: w.seq, ts: w.ts, event })
            }
            WireSessionEvent::QueryResult { id, .. } if id >= RESERVED_QUERY_ID_BASE => None,
            event => {
                let env = Envelope { session_id: w.session_id, seq: w.seq, ts: w.ts, event: event.into() };
                if matches!(
                    env.event,
                    SessionEventWire::Stream(crate::StreamEvent::Session(crate::SessionEvent::MessageHistory(_)))
                ) {
                    return None;
                }
                Some(env)
            }
        }
    }
}

#[async_trait::async_trait]
impl ClientTransport for SocketTransport {
    fn session_id(&self) -> &SessionId {
        &self.session
    }

    fn meta(&self) -> &SessionMeta {
        &self.meta
    }

    async fn send(&self, cmd: SessionCommand) -> Result<(), TransportError> {
        match self.writer.try_send(ClientFrame::Cmd { session_id: self.session.clone(), cmd }) {
            Ok(()) => Ok(()),
            Err(mpsc::error::TrySendError::Full(_)) => Err(TransportError::Backpressure),
            Err(mpsc::error::TrySendError::Closed(_)) => Err(TransportError::Closed),
        }
    }

    /// The daemon's conn stamps `from` with this connection's client id, so
    /// on the wire `send` and `send_from_self` are the same frame.
    async fn send_from_self(&self, cmd: SessionCommand) -> Result<(), TransportError> {
        self.send(cmd).await
    }

    async fn next_event(&mut self) -> Option<Envelope> {
        if self.ended {
            return None;
        }
        loop {
            match read_frame(&mut self.reader, &mut self.buf).await {
                Ok(Some(DaemonFrame::Event(w))) => {
                    let Some(env) = self.absorb(w) else { continue };
                    match &env.event {
                        SessionEventWire::SettingChanged(applied) => {
                            self.view.store(Arc::new(applied.view.clone()));
                        }
                        SessionEventWire::Ended { .. } => self.ended = true,
                        SessionEventWire::InputOwnerChanged { to, .. } => self.input_owner = *to,
                        SessionEventWire::Reloading { generation, .. } => {
                            self.reload_pending = Some(*generation);
                        }
                        other => self.roster.apply(other),
                    }
                    return Some(env);
                }
                Ok(Some(DaemonFrame::Error { message, .. })) => return Some(self.notice(message)),
                Ok(Some(DaemonFrame::Bye { reason })) => {
                    if let Some(ByeReason::Reloading { generation, .. }) = reason {
                        self.reload_pending = Some(generation);
                    }
                    self.ended = true;
                    if let Some(generation) = self.reload_pending {
                        self.last_error = Some(TransportError::Reloading { generation });
                    }
                    return None;
                }
                Ok(None) => {
                    self.ended = true;
                    if let Some(generation) = self.reload_pending {
                        self.last_error = Some(TransportError::Reloading { generation });
                    }
                    return None;
                }
                Ok(Some(other)) => {
                    tracing::debug!("socket_transport: unexpected frame after attach: {other:?}");
                    continue;
                }
                Err(e) => {
                    tracing::debug!("socket_transport: read error: {e}");
                    self.ended = true;
                    return None;
                }
            }
        }
    }

    fn view(&self) -> Arc<RuntimeView> {
        Arc::clone(&self.view.load())
    }

    fn client_id(&self) -> ClientId {
        self.client
    }

    fn mode(&self) -> AttachMode {
        self.mode
    }

    fn input_owner(&self) -> Option<ClientId> {
        self.input_owner
    }

    fn roster(&self) -> Roster {
        self.roster.clone()
    }

    fn client_input(&self) -> ClientInput {
        self.client_input
    }

    /// C3: after `Reloading`/EOF — backoff 100 ms ×2 → cap 5 s, total ≤
    /// `SYNAPS_TUI_ATTACH_RECONNECT_SECS` (60); `Hello{reconnect_of}` →
    /// `Attach::Existing{alias-resolved id, mode}` where `mode = Takeover`
    /// iff this client was the input owner (§2.7: two reconnecting mirrors
    /// cannot both take over). On success `self` IS the new connection; the
    /// returned snapshot is a full re-mirror.
    async fn reconnect(&mut self, mode: AttachMode) -> Result<AttachSnapshot, TransportError> {
        let was_owner = self.input_owner == Some(self.client);
        let generation = self.reload_pending.unwrap_or(self.welcome.generation);
        let mut hello = self.hello.clone();
        hello.reconnect_of = Some(ClientReconnect {
            previous_client: self.client,
            session_id: self.session.clone(),
            was_owner,
            generation,
        });
        let mode = if was_owner { AttachMode::Takeover } else { mode };
        let total = std::env::var("SYNAPS_TUI_ATTACH_RECONNECT_SECS")
            .ok()
            .and_then(|s| s.parse().ok())
            .map(Duration::from_secs)
            .unwrap_or(Duration::from_secs(60));
        let deadline = tokio::time::Instant::now() + total;
        let mut backoff = Duration::from_millis(100);
        let mut last;
        loop {
            match Connected::connect(&self.path, hello.clone()).await {
                Ok(conn) => {
                    match Self::attach(conn, Attach::Existing { session_id: self.session.clone(), mode }).await {
                        Ok((t, snap)) => {
                            *self = t;
                            return Ok(snap);
                        }
                        Err(TransportError::Refused(m)) => return Err(TransportError::Refused(m)),
                        Err(e) => last = e,
                    }
                }
                Err(TransportError::Version { client, daemon }) => {
                    return Err(TransportError::Version { client, daemon })
                }
                Err(e) => last = e,
            }
            if tokio::time::Instant::now() + backoff > deadline {
                return Err(last);
            }
            tokio::time::sleep(backoff).await;
            backoff = (backoff * 2).min(Duration::from_secs(5));
        }
    }

    async fn reconnect_once(&mut self, mode: AttachMode) -> Result<AttachSnapshot, TransportError> {
        // Delegate to the inherent method (same name, not trait-dispatched).
        SocketTransport::reconnect_once(self, mode).await
    }

    fn is_reload_pending(&self) -> bool {
        SocketTransport::is_reload_pending(self)
    }

    fn daemon_pid(&self) -> u32 {
        SocketTransport::daemon_pid(self)
    }

    fn daemon_binary(&self) -> Option<crate::session::transport::BinaryId> {
        Some(crate::session::transport::BinaryId::of(&self.welcome))
    }

    fn socket_path(&self) -> Option<&Path> {
        Some(&self.path)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ipc::Listener;
    use crate::{LlmEvent, StreamEvent};

    /// A hand-rolled daemon: Welcome, answer Ping, Attach → Attached, then
    /// echo every Cmd::Submit as an Event stream.
    async fn fake_daemon(mut listener: Listener, protocol_version: u32) {
        let stream = listener.accept().await.unwrap();
        let (r, mut w) = crate::ipc::split_server(stream);
        let mut reader = BufReader::new(r);
        let mut line = String::new();
        let mut seq = 0u64;
        let sid = SessionId::from("fake-1");
        loop {
            line.clear();
            if reader.read_line(&mut line).await.unwrap() == 0 {
                break;
            }
            let frame: ClientFrame = decode_line(line.trim_end()).unwrap();
            let reply = match frame {
                ClientFrame::Hello(h) => {
                    if h.protocol_version != protocol_version {
                        let f = DaemonFrame::Refused {
                            reason: RefuseReason::Version { daemon_version: protocol_version, min: protocol_version, max: protocol_version },
                            message: "version".into(),
                        };
                        w.write_all(encode_line(&f).unwrap().as_bytes()).await.unwrap();
                        break;
                    }
                    DaemonFrame::Welcome(Welcome {
                        protocol_version,
                        daemon_version: "t".into(),
                        pid: 1,
                        profile: None,
                        sessions: vec![],
                        progressive_tool_disclosure: true,
                        generation: 1,
                        daemon_build: String::new(),
                    })
                }
                ClientFrame::Ping => DaemonFrame::Pong { pid: 1, uptime_s: 0, sessions: 0 },
                ClientFrame::Attach(_) => DaemonFrame::Attached(AttachedWire::new(
                    ClientId(7),
                    AttachSnapshot {
                        meta: crate::session::handle::echo::meta_for(&sid),
                        view: crate::session::handle::echo::view_for(),
                        conversation: ConversationSnapshot::default(),
                        streaming: false,
                        replay: vec![],
                        pending_prompts: vec![],
                        clients: vec![],
                        roster: Vec::new(),
                        client_input: ClientInput::wire_default(),
                        input_owner: Some(ClientId(7)),
                        display_tail: None,
                    },
                )),
                ClientFrame::Cmd { cmd: SessionCommand::Submit { text, .. }, .. } => {
                    seq += 1;
                    DaemonFrame::Event(WireEnvelope {
                        session_id: sid.clone(),
                        seq,
                        ts: chrono::Utc::now(),
                        event: SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(text))).into(),
                    })
                }
                ClientFrame::Cmd { cmd: SessionCommand::Set { id, .. }, .. } => {
                    seq += 1;
                    let mut view = crate::session::handle::echo::view_for();
                    view.model = "changed".into();
                    DaemonFrame::Event(WireEnvelope {
                        session_id: sid.clone(),
                        seq,
                        ts: chrono::Utc::now(),
                        event: SessionEventWire::SettingChanged(SettingApplied {
                            id,
                            setting: "model".into(),
                            ok: true,
                            message: None,
                            view,
                            clamp: None,
                        })
                        .into(),
                    })
                }
                // Checkpoint = "the daemon is about to reload": Reloading event, then Bye{Reloading}.
                ClientFrame::Cmd { cmd: SessionCommand::Checkpoint { .. }, .. } => {
                    seq += 1;
                    let ev = DaemonFrame::Event(WireEnvelope {
                        session_id: sid.clone(),
                        seq,
                        ts: chrono::Utc::now(),
                        event: WireSessionEvent::Reloading { generation: 2, retry_after_ms: 500 },
                    });
                    w.write_all(encode_line(&ev).unwrap().as_bytes()).await.unwrap();
                    let bye = DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation: 2, retry_after_ms: 500 }) };
                    w.write_all(encode_line(&bye).unwrap().as_bytes()).await.unwrap();
                    break;
                }
                // KeepWarm = ownership change notice.
                ClientFrame::Cmd { cmd: SessionCommand::KeepWarm { .. }, .. } => {
                    seq += 1;
                    DaemonFrame::Event(WireEnvelope {
                        session_id: sid.clone(),
                        seq,
                        ts: chrono::Utc::now(),
                        event: WireSessionEvent::InputOwnerChanged {
                            from: Some(ClientId(7)),
                            to: Some(ClientId(9)),
                            reason: OwnerChangeReason::Takeover,
                        },
                    })
                }
                // Steer = "the daemon's history drifted": digest for [user:text] without a MessageHistory.
                ClientFrame::Cmd { cmd: SessionCommand::Steer { text }, .. } => {
                    seq += 1;
                    let snap = ConversationSnapshot {
                        api_messages: vec![Arc::new(serde_json::json!({"role":"user","content":text}))],
                        cost: 1.5,
                        ..Default::default()
                    };
                    DaemonFrame::Event(WireEnvelope {
                        session_id: sid.clone(),
                        seq,
                        ts: chrono::Utc::now(),
                        event: SessionEventWire::Conversation(snap).into(),
                    })
                }
                ClientFrame::Cmd { cmd: SessionCommand::Query { id, query: SessionQuery::Messages }, .. } => {
                    seq += 1;
                    DaemonFrame::Event(WireEnvelope {
                        session_id: sid.clone(),
                        seq,
                        ts: chrono::Utc::now(),
                        event: WireSessionEvent::QueryResult {
                            id,
                            value: serde_json::json!([{"role":"user","content":"drifted"}]),
                        },
                    })
                }
                ClientFrame::Bye => {
                    let _ = w.write_all(encode_line(&DaemonFrame::Bye { reason: None }).unwrap().as_bytes()).await;
                    break;
                }
                ClientFrame::Cmd { cmd: SessionCommand::Detach { .. }, .. } => continue,
                _ => DaemonFrame::Error { session_id: None, message: "nope".into() },
            };
            let bytes = encode_line(&reply).unwrap();
            // `split:` Submit text: the frame arrives in two writes 50 ms
            // apart (a frame split across reads, as a large one can be).
            if bytes.contains("split:") {
                let (a, b) = bytes.as_bytes().split_at(bytes.len() / 2);
                w.write_all(a).await.unwrap();
                w.flush().await.unwrap();
                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                w.write_all(b).await.unwrap();
                continue;
            }
            w.write_all(bytes.as_bytes()).await.unwrap();
        }
    }

    fn sock() -> (tempfile::TempDir, std::path::PathBuf) {
        let d = tempfile::tempdir().unwrap();
        let p = d.path().join("d.sock");
        (d, p)
    }

    /// A daemon built before `AuthState` decodes the frame into a
    /// `ClientFrame` without that variant, fails, and answers
    /// `format!("malformed frame: {e}")` (daemon/conn.rs `Read::Bad`). That
    /// reply, and no other error, means "the daemon predates auth state".
    #[test]
    fn predates_auth_state_matches_an_old_daemons_decode_error_and_nothing_else() {
        #[derive(serde::Deserialize)]
        #[serde(tag = "type", rename_all = "snake_case")]
        #[allow(dead_code, clippy::large_enum_variant)]
        enum OldClientFrame {
            Hello(Hello),
            Ping,
            Sessions,
            Shutdown {
                #[serde(default)]
                force: bool,
            },
            Purge,
            Bye,
        }
        for query in [
            agent_core::auth::AuthStateQuery::default(),
            agent_core::auth::AuthStateQuery { provider: Some("anthropic".into()), account: Some("b".into()), clear: true },
        ] {
            let line = encode_line(&ClientFrame::AuthState(query)).unwrap();
            let err = decode_line::<OldClientFrame>(line.trim_end()).err().expect("an old daemon cannot decode it");
            let reply = format!("malformed frame: {err}");
            assert!(predates_auth_state(&reply), "{reply}");
        }
        for other in [
            "auth state: unknown account `zzz`",
            "malformed frame: expected value at line 1 column 1",
            "malformed frame: unknown variant `auth_status`, expected one of `hello`, `ping`",
            "malformed frame: missing field `clear`",
            "control frames are not accepted on an attached connection",
            "duplicate hello",
            "",
        ] {
            assert!(!predates_auth_state(other), "{other:?}");
        }
    }

    /// The TUI polls `next_event()` inside `tokio::select!` next to a 16 ms
    /// tick, so the future is dropped whenever the tick wins. A frame split
    /// across reads must survive that cancellation.
    #[tokio::test]
    async fn next_event_survives_cancellation_mid_frame() {
        let (_d, path) = sock();
        let l = Listener::bind(&path).unwrap();
        let srv = tokio::spawn(fake_daemon(l, PROTOCOL_VERSION));
        let conn = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.unwrap();
        let (mut t, _) = SocketTransport::attach(conn, Attach::Existing { session_id: "fake-1".into(), mode: AttachMode::Mirror })
            .await
            .unwrap();
        t.send(SessionCommand::Submit { text: "split:hello".into(), attachments: vec![] }).await.unwrap();
        // First half is in; the tick wins and drops the read.
        tokio::select! {
            e = t.next_event() => panic!("frame cannot be complete yet: {e:?}"),
            _ = tokio::time::sleep(std::time::Duration::from_millis(20)) => {}
        }
        let e = tokio::time::timeout(std::time::Duration::from_secs(2), t.next_event())
            .await
            .expect("no event");
        match e.map(|e| e.event) {
            Some(SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(s)))) => assert_eq!(s, "split:hello"),
            other => panic!("frame lost to cancellation: {other:?} (last_error={:?})", t.last_error()),
        }
        t.detach().await;
        let _ = srv.await;
    }

    #[tokio::test]
    async fn socket_transport_handshake_attach_pump() {
        let (_d, path) = sock();
        let l = Listener::bind(&path).unwrap();
        let srv = tokio::spawn(fake_daemon(l, PROTOCOL_VERSION));

        let mut conn = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.unwrap();
        assert_eq!(conn.welcome.protocol_version, PROTOCOL_VERSION);
        assert_eq!(conn.ping().await.unwrap().pid, 1);
        let (mut t, snap) = SocketTransport::attach(conn, Attach::Existing { session_id: "fake-1".into(), mode: AttachMode::Mirror })
            .await
            .unwrap();
        assert_eq!(t.client_id(), ClientId(7));
        assert_eq!(t.mode(), AttachMode::Mirror);
        assert_eq!(t.input_owner(), Some(ClientId(7)));
        assert_eq!(snap.meta.id.as_str(), "fake-1");
        assert_eq!(t.session_id().as_str(), "fake-1");
        assert_eq!(t.view().model, "echo");

        t.send(SessionCommand::Submit { text: "hello".into(), attachments: vec![] }).await.unwrap();
        let e = t.next_event().await.unwrap();
        match e.event {
            SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::Text(s))) => assert_eq!(s, "hello"),
            other => panic!("{other:?}"),
        }
        t.send_from_self(SessionCommand::Set { id: 11, setting: SessionSetting::Model { model: "changed".into() } })
            .await
            .unwrap();
        let e = t.next_event().await.unwrap();
        assert!(matches!(e.event, SessionEventWire::SettingChanged(SettingApplied { id: 11, .. })));
        assert_eq!(t.view().model, "changed");
        t.send(SessionCommand::KeepWarm { on: true }).await.unwrap();
        let e = t.next_event().await.unwrap();
        assert!(matches!(e.event, SessionEventWire::InputOwnerChanged { .. }));
        assert_eq!(t.input_owner(), Some(ClientId(9)));

        // Error → SystemNotice; Bye → None.
        t.send(SessionCommand::Save).await.unwrap();
        assert!(matches!(t.next_event().await.unwrap().event, SessionEventWire::SystemNotice(_)));
        t.detach().await;
        assert!(t.next_event().await.is_none());
        assert!(t.next_event().await.is_none());
        srv.await.unwrap();
    }

    #[tokio::test]
    async fn conversation_digest_miss_requeries_messages() {
        let (_d, path) = sock();
        let l = Listener::bind(&path).unwrap();
        let srv = tokio::spawn(fake_daemon(l, PROTOCOL_VERSION));
        let conn = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.unwrap();
        let (mut t, _) = SocketTransport::attach(conn, Attach::Existing { session_id: "fake-1".into(), mode: AttachMode::Mirror })
            .await
            .unwrap();
        assert!(t.messages().is_empty());
        t.send(SessionCommand::Steer { text: "drifted".into() }).await.unwrap();
        // The digest misses the (empty) mirror → transport re-queries → one
        // Conversation with the fetched history and the digest's cost.
        let e = t.next_event().await.unwrap();
        match e.event {
            SessionEventWire::Conversation(c) => {
                assert_eq!(c.api_messages.len(), 1);
                assert_eq!(c.api_messages[0]["content"], "drifted");
                assert_eq!(c.cost, 1.5);
            }
            other => panic!("{other:?}"),
        }
        assert_eq!(t.messages().len(), 1);
        // Same digest again now matches the mirror: no re-query, served locally.
        t.send(SessionCommand::Steer { text: "drifted".into() }).await.unwrap();
        let e = t.next_event().await.unwrap();
        assert!(matches!(e.event, SessionEventWire::Conversation(ref c) if c.api_messages.len() == 1));
        t.detach().await;
        assert!(t.next_event().await.is_none());
        srv.await.unwrap();
    }

    #[tokio::test]
    async fn reloading_then_bye_sets_last_error_and_reconnect_gives_up_on_a_dead_socket() {
        let (_d, path) = sock();
        let l = Listener::bind(&path).unwrap();
        let srv = tokio::spawn(fake_daemon(l, PROTOCOL_VERSION));
        let conn = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.unwrap();
        let (mut t, _) = SocketTransport::attach(conn, Attach::Existing { session_id: "fake-1".into(), mode: AttachMode::Observe })
            .await
            .unwrap();
        assert_eq!(t.mode(), AttachMode::Observe);
        assert!(t.last_error().is_none());
        t.send(SessionCommand::Checkpoint { reason: CheckpointReason::Reload }).await.unwrap();
        let e = t.next_event().await.unwrap();
        assert!(matches!(e.event, SessionEventWire::Reloading { generation: 2, .. }));
        assert!(t.next_event().await.is_none());
        assert!(matches!(t.last_error(), Some(TransportError::Reloading { generation: 2 })));
        srv.await.unwrap();
        // Nobody listens any more: the backoff loop gives up at the budget.
        std::env::set_var("SYNAPS_TUI_ATTACH_RECONNECT_SECS", "1");
        let r = t.reconnect(AttachMode::Observe).await;
        std::env::remove_var("SYNAPS_TUI_ATTACH_RECONNECT_SECS");
        assert!(r.is_err());
    }

    #[tokio::test]
    async fn attach_error_frame_is_refused() {
        let (_d, path) = sock();
        let mut l = Listener::bind(&path).unwrap();
        // A daemon that welcomes, then answers the Attach with an Error frame.
        let srv = tokio::spawn(async move {
            let stream = l.accept().await.unwrap();
            let (r, mut w) = crate::ipc::split_server(stream);
            let mut reader = BufReader::new(r);
            let mut line = String::new();
            reader.read_line(&mut line).await.unwrap();
            let welcome = DaemonFrame::Welcome(Welcome {
                protocol_version: PROTOCOL_VERSION,
                daemon_version: "t".into(),
                pid: 1,
                profile: None,
                sessions: vec![],
                progressive_tool_disclosure: true,
                generation: 1,
                daemon_build: String::new(),
            });
            w.write_all(encode_line(&welcome).unwrap().as_bytes()).await.unwrap();
            line.clear();
            reader.read_line(&mut line).await.unwrap();
            let err = DaemonFrame::Error { session_id: None, message: "unknown session".into() };
            w.write_all(encode_line(&err).unwrap().as_bytes()).await.unwrap();
        });
        let conn = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.unwrap();
        let err = SocketTransport::attach(conn, Attach::Existing { session_id: "nope".into(), mode: AttachMode::Mirror })
            .await
            .err()
            .unwrap();
        assert!(matches!(err, TransportError::Refused(ref m) if m == "unknown session"), "{err}");
        srv.await.unwrap();
    }

    #[tokio::test]
    async fn socket_transport_refuses_version_mismatch() {
        let (_d, path) = sock();
        let l = Listener::bind(&path).unwrap();
        let srv = tokio::spawn(fake_daemon(l, 99));
        let err = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.err().unwrap();
        assert!(matches!(err, TransportError::Version { client: PROTOCOL_VERSION, daemon: 99 }), "{err}");
        srv.await.unwrap();
    }

    /// An older daemon answers `Close` with serde's unknown-variant error;
    /// the client reports "too old" (exit 2), not a generic failure.
    #[test]
    fn close_maps_each_daemon_answer_and_an_old_daemons_unknown_variant_error() {
        let old = DaemonFrame::Error {
            session_id: None,
            message: "malformed frame: unknown variant `close`, expected one of `hello`, `ping`".into(),
        };
        let err = close_reply(Some(old), "0.9.0").unwrap_err();
        assert!(matches!(err, TransportError::DaemonTooOld { ref daemon_version, .. } if daemon_version == "0.9.0"), "{err}");
        assert!(err.to_string().contains("synaps daemon reload"), "{err}");

        let closed = DaemonFrame::Closed { session_id: "s".into(), text: "closing session s".into() };
        assert_eq!(close_reply(Some(closed), "1").unwrap(), "closing session s");
        let busy = DaemonFrame::Refused { reason: RefuseReason::Busy, message: "not closed: a turn is running".into() };
        assert!(matches!(close_reply(Some(busy), "1"), Err(TransportError::Refused(ref m)) if m.contains("turn")));
        let unknown = DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: "no live session s".into() };
        assert!(matches!(close_reply(Some(unknown), "1"), Err(TransportError::UnknownSession(_))));
        let other = DaemonFrame::Error { session_id: None, message: "close: no reply".into() };
        assert!(matches!(close_reply(Some(other), "1"), Err(TransportError::Protocol(_))));
        assert!(matches!(close_reply(None, "1"), Err(TransportError::Closed)));
    }

    #[test]
    fn pin_maps_each_daemon_answer_and_an_old_daemons_unknown_variant_error() {
        let old = DaemonFrame::Error {
            session_id: None,
            message: "malformed frame: unknown variant `pin`, expected one of `hello`, `ping`".into(),
        };
        let err = pin_reply(Some(old), "0.9.0").unwrap_err();
        assert!(
            matches!(err, TransportError::DaemonTooOld { ref daemon_version, request: "daemon pin" } if daemon_version == "0.9.0"),
            "{err}"
        );
        assert!(err.to_string().contains("synaps daemon reload"), "{err}");

        assert!(pin_reply(Some(DaemonFrame::Pinned { session_id: "s".into(), on: true }), "1").unwrap());
        assert!(!pin_reply(Some(DaemonFrame::Pinned { session_id: "s".into(), on: false }), "1").unwrap());
        let busy = DaemonFrame::Refused { reason: RefuseReason::Busy, message: "daemon reloading".into() };
        assert!(matches!(pin_reply(Some(busy), "1"), Err(TransportError::Refused(ref m)) if m.contains("reloading")));
        let unknown = DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: "no live session s".into() };
        assert!(matches!(pin_reply(Some(unknown), "1"), Err(TransportError::UnknownSession(_))));
        let other = DaemonFrame::Error { session_id: None, message: "pin: timed out".into() };
        assert!(matches!(pin_reply(Some(other), "1"), Err(TransportError::Protocol(_))));
        let closed = DaemonFrame::Closed { session_id: "s".into(), text: "x".into() };
        assert!(matches!(pin_reply(Some(closed), "1"), Err(TransportError::Protocol(_))));
        assert!(matches!(pin_reply(None, "1"), Err(TransportError::Closed)));
    }

    #[test]
    fn rename_maps_each_daemon_answer_and_an_old_daemons_unknown_variant_error() {
        use crate::session::wire::RenameRefusal;
        let old = DaemonFrame::Error {
            session_id: None,
            message: "malformed frame: unknown variant `rename`, expected one of `hello`, `ping`".into(),
        };
        let err = rename_reply(Some(old), "0.9.0").unwrap_err();
        assert!(
            matches!(err, TransportError::DaemonTooOld { ref daemon_version, request: "daemon rename" } if daemon_version == "0.9.0"),
            "{err}"
        );
        assert!(err.to_string().contains("predates `daemon rename`"), "{err}");

        let ok = DaemonFrame::Renamed { session_id: "s".into(), name: Some("axon".into()), in_memory_only: false };
        assert_eq!(rename_reply(Some(ok), "1").unwrap(), Ok(Renamed { name: Some("axon".into()), in_memory_only: false }));
        let cleared = DaemonFrame::Renamed { session_id: "s".into(), name: None, in_memory_only: true };
        assert_eq!(rename_reply(Some(cleared), "1").unwrap(), Ok(Renamed { name: None, in_memory_only: true }));
        let taken = DaemonFrame::RenameFailed {
            session_id: "s".into(),
            reason: RenameRefusal::Taken,
            message: "name 'axon' already used by session t".into(),
            taken_by: Some("t".into()),
            current_name: None,
        };
        let failure = rename_reply(Some(taken), "1").unwrap().unwrap_err();
        assert_eq!((failure.reason, failure.taken_by.as_deref()), (RenameRefusal::Taken, Some("t")));
        let busy = DaemonFrame::Refused { reason: RefuseReason::Busy, message: "daemon reloading".into() };
        assert!(matches!(rename_reply(Some(busy), "1"), Err(TransportError::Refused(_))));
        let unknown = DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: "no live session s".into() };
        assert!(matches!(rename_reply(Some(unknown), "1"), Err(TransportError::UnknownSession(_))));
        let other = DaemonFrame::Error { session_id: None, message: "rename: no reply".into() };
        assert!(matches!(rename_reply(Some(other), "1"), Err(TransportError::Protocol(_))));
        assert!(matches!(rename_reply(None, "1"), Err(TransportError::Closed)));
    }

    #[test]
    fn reload_extension_maps_each_daemon_answer_and_an_old_daemons_unknown_variant_error() {
        let old = DaemonFrame::Error {
            session_id: None,
            message: "malformed frame: unknown variant `reload_extension`, expected one of `hello`, `ping`".into(),
        };
        let err = reload_extension_reply(Some(old), "0.9.0").unwrap_err();
        assert!(
            matches!(err, TransportError::DaemonTooOld { ref daemon_version, request: "daemon reload-extension" } if daemon_version == "0.9.0"),
            "{err}"
        );

        let report = ExtensionReloadReport {
            extension: "axel".into(),
            was_loaded: true,
            deferred: false,
            tools: vec!["axel:axel_search".into()],
            removed_tools: vec![],
            plugin_dir: std::path::PathBuf::from("/p/axel"),
        };
        let ok = DaemonFrame::ExtensionReloaded { report: report.clone() };
        assert_eq!(reload_extension_reply(Some(ok), "1").unwrap(), Ok(report));
        let failed = DaemonFrame::ExtensionReloadFailed { extension: "axel".into(), message: "spawn failed".into(), unloaded: true };
        assert_eq!(
            reload_extension_reply(Some(failed), "1").unwrap(),
            Err(ExtensionReloadFailure { message: "spawn failed".into(), unloaded: true })
        );
        let busy = DaemonFrame::Refused { reason: RefuseReason::Busy, message: "daemon reloading".into() };
        assert!(matches!(reload_extension_reply(Some(busy), "1"), Err(TransportError::Refused(ref m)) if m.contains("reloading")));
        // A plain `Error` (not the unknown-variant one) stays a protocol error.
        let other = DaemonFrame::Error { session_id: None, message: "frame too large".into() };
        assert!(matches!(reload_extension_reply(Some(other), "1"), Err(TransportError::Protocol(_))));
        let pinned = DaemonFrame::Pinned { session_id: "s".into(), on: true };
        assert!(matches!(reload_extension_reply(Some(pinned), "1"), Err(TransportError::Protocol(_))));
        assert!(matches!(reload_extension_reply(None, "1"), Err(TransportError::Closed)));
    }

    #[tokio::test]
    async fn socket_transport_connect_refused_when_nobody_listens() {
        let (_d, path) = sock();
        let err = SocketTransport::connect(&path, Hello::new(ClientKind::Test)).await.err().unwrap();
        assert!(matches!(err, TransportError::Io(_)));
    }
}