agent_engine/session/

wire.rs

//! Wire frames for the daemon UDS (PLAN-phase2 §2.9, S289 D-2 amended).
//!
//! Framing is `synaps rpc`'s: one JSON object per `\n`, at most
//! [`DAEMON_MAX_FRAME_BYTES`] (64 MiB — distinct from rpc's 1 MiB because
//! `Attached` carries the whole `api_messages`). Enforced symmetrically:
//! `encode_line` refuses to build an oversize frame, `decode_line` and both
//! readers refuse to accept one. `WireSessionEvent` is a serde mirror of
//! `SessionEventWire` — `StreamEvent`/`TurnError`/`ExtensionLoaderEvent` are
//! not `Serialize`, so they are mirrored variant-for-variant here and
//! converted only at the socket boundary. In-process transports never touch
//! this module.
//!
//! The one *lossy* mirror is `Conversation`: on the wire it is a
//! [`ConversationDigest`] (len + hash + tokens/cost), never the messages.
//! Full history travels only in `Attached` and `QueryResult{Messages}`;
//! `SocketTransport` rebuilds the snapshot from its local mirror
//! (`Attached` + `Stream(MessageHistory)`) and re-queries on a hash miss.
//!
//! Security: nothing in here carries a credential. `Welcome`/`Attached` are
//! summaries; `ClientFrame`'s `Debug` redacts `Answer.value` and
//! `Submit.text` so a frame can be traced by *type* without leaking bodies.

use std::path::PathBuf;

use serde::{Deserialize, Serialize};

use super::types::*;
use super::view::RuntimeView;
use crate::{AgentEvent, LlmEvent, SessionEvent, StreamEvent, TurnError, TurnOutcome};

/// Separate from `RPC_PROTOCOL_VERSION`; bump on any non-additive change.
/// v2 (phase 3): `Set` became a struct variant, `Bye` gained a reason.
/// v3 (E-P0): driver events (`DriverArmed`, `DriverRevoked`,
/// `DriverTurnOutcome`); old clients skip unknown via `#[serde(other)]`.
/// Protocol stays exact-match; only *binary* versions compare directionally
/// (reload, §2.8).
/// `Close`/`Closed` were added without a bump: an older daemon answers the
/// unknown `ClientFrame` with `Error{"malformed frame: unknown variant
/// `close`…"}` (the client maps that to "daemon too old"), whereas a bump
/// would refuse every older TUI that reconnects after a `daemon reload`.
/// Additive `ClientFrame` variants that only a new client sends follow this
/// rule; `Closed` only ever answers `Close`, so old clients never see it.
/// `Pin`/`Pinned` (`synaps daemon pin`) follow the same rule, and so does
/// `SessionMeta.pinned` (`#[serde(default)]`: an older daemon omits it).
/// `ReloadExtension` and its two replies (`ExtensionReloaded`,
/// `ExtensionReloadFailed`) follow it too.
pub const PROTOCOL_VERSION: u32 = 3;
/// Exact-match policy today (`min == max == PROTOCOL_VERSION`).
pub const PROTOCOL_MIN: u32 = PROTOCOL_VERSION;
pub const PROTOCOL_MAX: u32 = PROTOCOL_VERSION;
/// rpc's cap, kept for reference; daemon frames use [`DAEMON_MAX_FRAME_BYTES`].
pub const RPC_MAX_FRAME_BYTES: usize = agent_core::core::rpc_dispatch::MAX_FRAME_BYTES;
/// Hard cap on one daemon frame, both directions (`Attached` ships the
/// whole conversation; 64 MiB is well past any context window's JSON).
pub const DAEMON_MAX_FRAME_BYTES: usize = 64 * 1024 * 1024;
/// Alias used by the readers/writers of this protocol.
pub const MAX_FRAME_BYTES: usize = DAEMON_MAX_FRAME_BYTES;
/// Query ids at or above this are reserved for the transport/daemon and
/// never surfaced to a client as `QueryResult`.
pub const RESERVED_QUERY_ID_BASE: u64 = 1 << 63;
/// `SocketTransport` re-fetches `api_messages` under this id on a digest miss.
pub const DIGEST_RESYNC_QUERY_ID: u64 = RESERVED_QUERY_ID_BASE;
/// The daemon's idle monitor probes `Status` under this id.
pub const IDLE_PROBE_QUERY_ID: u64 = RESERVED_QUERY_ID_BASE + 1;
/// `Checkpoint` replies `QueryResult{id: CHECKPOINT_QUERY_ID, {ok:true}}`
/// so `reload.rs` can await it per session.
pub const CHECKPOINT_QUERY_ID: u64 = RESERVED_QUERY_ID_BASE + 2;
/// `SettingChanged{id}` broadcast when an engine command (`/model`,
/// `/thinking`, …) changed the view: not a reply to any client's `Set`
/// (client ids never reach the reserved range), only the new view, so every
/// attached client mirrors it, not just the one that typed the command.
pub const VIEW_CHANGED_SETTING_ID: u64 = RESERVED_QUERY_ID_BASE + 3;
/// `synaps daemon close` sends the engine command `close` under a fresh id
/// at or above this one per request (`daemon/close.rs`) and awaits that id's
/// `QueryResult`, so concurrent closes never take each other's reply.
/// Attached clients never see these.
pub const CLOSE_QUERY_ID_BASE: u64 = RESERVED_QUERY_ID_BASE + (1 << 32);

/// The binary version the daemon/client report in the handshake.
pub fn binary_version() -> String {
    env!("CARGO_PKG_VERSION").to_string()
}

static BUILD_ID: std::sync::OnceLock<String> = std::sync::OnceLock::new();

/// Record this process's build id. The `synaps` binary calls it first thing
/// in `main` with the id its build script baked in (the short commit,
/// `-dirty` with uncommitted changes, or `SYNAPS_BUILD_ID` from the build
/// environment). It lives in the root crate, not here, so a new commit does
/// not recompile the engine. First call wins; later calls are ignored.
pub fn set_build_id(id: &str) {
    let _ = BUILD_ID.set(id.trim().to_string());
}

/// Which build of [`binary_version`] this is, as set by [`set_build_id`];
/// empty when unknown (never set: library embedders, tests, packaged builds
/// without git). Two builds of one release share the version, not this.
pub fn build_id() -> &'static str {
    BUILD_ID.get().map(String::as_str).unwrap_or("")
}

// ── client → daemon ───────────────────────────────────────────────────────────

/// Client → daemon. First frame on a connection MUST be `Hello`.
#[derive(Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ClientFrame {
    Hello(Hello),
    // ── lightweight control fast path: no session allocated, no Attach ──
    Ping,
    Sessions,
    Shutdown {
        #[serde(default)]
        force: bool,
    },
    // ── session-scoped ──
    Attach(Attach),
    Cmd {
        session_id: SessionId,
        cmd: SessionCommand,
    },
    /// Control fast path (no session): checkpoint every session, write
    /// reload-state, execv self. Reply: `Reloading` to every attached
    /// client, `Bye` to this one. (C3)
    Reload {
        #[serde(default)]
        now: bool,
        #[serde(default)]
        drain_secs: Option<u64>,
        #[serde(default)]
        exe: Option<PathBuf>,
    },
    /// Control fast path: `memstat::purge_arenas()` in the daemon (bench
    /// hygiene). Reply: `Pong`. (C2)
    Purge,
    /// Control fast path: end one session as `/close [force]` would, with
    /// host authority (no attach, never unparks). A session with clients
    /// attached is refused unless `attached` (`--attached`); `force` only
    /// overrides work in flight. Reply: `Closed`, or
    /// `Refused{Busy|UnknownSession}`. Not accepted on an attached
    /// connection (that client has `/close`).
    Close {
        session_id: SessionId,
        #[serde(default)]
        force: bool,
        #[serde(default)]
        attached: bool,
    },
    /// Control fast path: pin (`on`) or unpin one session, as `/pin on|off`
    /// would, with host authority. `on` is required so a frame always says
    /// which way it goes. Pinning a parked session unparks it; unpinning one
    /// leaves it parked. Reply: `Pinned`, or `Refused{Busy|UnknownSession}`.
    /// Not accepted on an attached connection (that client has `/pin`).
    Pin {
        session_id: SessionId,
        on: bool,
    },
    /// Control fast path (no session): the in-memory account state of the
    /// broker every session of this daemon shares (`synaps auth state`),
    /// cleared first when `clear` is set. Reply: `AuthState`. A daemon that
    /// predates this frame answers `Error` "malformed frame: unknown variant
    /// `auth_state`, ..." and keeps the connection usable.
    AuthState(agent_core::auth::AuthStateQuery),
    /// Control fast path (no session): stop one extension and start it
    /// again from what is installed now (`synaps daemon reload-extension`).
    /// Extensions are daemon-wide, so every session gets the new version;
    /// no session is checkpointed and no turn is cancelled (only a call
    /// into the old process that is in flight fails). Also loads a plugin
    /// installed after the daemon started. Reply: `ExtensionReloaded`,
    /// `ExtensionReloadFailed`, or `Refused{Busy}` while the daemon
    /// reloads. Not accepted on an attached connection.
    ReloadExtension {
        extension: String,
    },
    Bye,
}

impl std::fmt::Debug for ClientFrame {
    /// Frame *types* only for anything that can carry user text or a secret
    /// (`SessionCommand`'s own `Debug` redacts bodies without lengths).
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Hello(h) => f.debug_tuple("Hello").field(h).finish(),
            Self::Ping => f.write_str("Ping"),
            Self::Sessions => f.write_str("Sessions"),
            Self::Shutdown { force } => f.debug_struct("Shutdown").field("force", force).finish(),
            Self::Attach(a) => f.debug_tuple("Attach").field(a).finish(),
            Self::Cmd { session_id, cmd } => {
                f.debug_struct("Cmd").field("session_id", session_id).field("cmd", cmd).finish()
            }
            Self::Reload { now, drain_secs, exe } => f
                .debug_struct("Reload")
                .field("now", now)
                .field("drain_secs", drain_secs)
                .field("exe", exe)
                .finish(),
            Self::Purge => f.write_str("Purge"),
            Self::Close { session_id, force, attached } => f
                .debug_struct("Close")
                .field("session_id", session_id)
                .field("force", force)
                .field("attached", attached)
                .finish(),
            Self::Pin { session_id, on } => f
                .debug_struct("Pin")
                .field("session_id", session_id)
                .field("on", on)
                .finish(),
            Self::AuthState(query) => f.debug_tuple("AuthState").field(query).finish(),
            Self::ReloadExtension { extension } => f
                .debug_struct("ReloadExtension")
                .field("extension", extension)
                .finish(),
            Self::Bye => f.write_str("Bye"),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Hello {
    pub protocol_version: u32,
    pub client: ClientMeta,
    pub cwd: PathBuf,
    pub client_version: String,
    /// Client environment snapshot (secrets and `SYNAPS_CLIENT_*`/`SYNAPS_TUI_*`/
    /// `SYNAPS_DAEMON_*`/`SYNAPS_MEM_TRACE*` stripped). `None` = in-process
    /// (inherit process env). Legacy frames without this field decode as `None`.
    #[serde(default)]
    pub env: Option<super::types::SessionEnv>,
    /// Names of env vars stripped as secrets at the client (T5). Names only,
    /// never values. `serde(default)` → legacy frames decode as empty.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub env_stripped: Vec<String>,
    /// Set when re-connecting after `Reloading`/EOF (C3/A4).
    #[serde(default)]
    pub reconnect_of: Option<ClientReconnect>,
}

/// Who this connection used to be, so the daemon can restore ownership by
/// reconnect order + `was_owner`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClientReconnect {
    pub previous_client: ClientId,
    pub session_id: SessionId,
    pub was_owner: bool,
    pub generation: u64,
}

impl Hello {
    pub fn new(kind: ClientKind) -> Self {
        let (env, env_stripped) = super::types::capture_client_env();
        Self {
            protocol_version: PROTOCOL_VERSION,
            client: ClientMeta {
                terminal: std::env::var("TERM").ok(),
                ..ClientMeta::new(kind)
            },
            cwd: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
            client_version: binary_version(),
            env: Some(env),
            env_stripped,
            reconnect_of: None,
        }
    }

    /// Ask to be called `name` in the session ([`ClientMeta::name`]).
    pub fn with_name(mut self, name: Option<String>) -> Self {
        self.client.name = name;
        self
    }

    /// Ask for `Full` (mirror) or `Digest` (digest + display tail) history.
    pub fn with_history(mut self, history: HistoryMode) -> Self {
        self.client.history = history;
        self
    }

    /// Display items wanted in `Attached.display_tail` / `DisplayTail` queries.
    pub fn with_tail_items(mut self, tail_items: usize) -> Self {
        self.client.tail_items = tail_items;
        self
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "attach", rename_all = "snake_case")]
pub enum Attach {
    Existing {
        session_id: SessionId,
        #[serde(default = "mirror")]
        mode: AttachMode,
    },
    Create {
        #[serde(default)]
        config: SessionConfig,
        #[serde(default = "mirror")]
        mode: AttachMode,
    },
}

fn mirror() -> AttachMode {
    AttachMode::Mirror
}

// ── daemon → client ───────────────────────────────────────────────────────────

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
#[allow(clippy::large_enum_variant)]
pub enum DaemonFrame {
    Welcome(Welcome),
    /// Then the daemon closes.
    Refused {
        reason: RefuseReason,
        message: String,
    },
    Pong {
        pid: u32,
        uptime_s: u64,
        sessions: usize,
    },
    SessionList {
        sessions: Vec<SessionMeta>,
    },
    Attached(AttachedWire),
    Event(WireEnvelope),
    Error {
        session_id: Option<SessionId>,
        message: String,
    },
    /// Reply to `Close`: the session ended; `text` is the actor's `/close`
    /// reply (how to continue it).
    Closed {
        session_id: SessionId,
        text: String,
    },
    /// Reply to `Pin`: the session's pin is now `on`.
    Pinned {
        session_id: SessionId,
        on: bool,
    },
    /// Reply to `ReloadExtension`: the new version is loaded.
    ExtensionReloaded {
        report: crate::extensions::manager::ExtensionReloadReport,
    },
    /// Reply to `ReloadExtension`: it was not reloaded. `unloaded`: the old
    /// instance was stopped before the new one failed to start, so the
    /// extension is not loaded now; `false` = the running one (if any) was
    /// not touched.
    ExtensionReloadFailed {
        extension: String,
        message: String,
        #[serde(default)]
        unloaded: bool,
    },
    /// `Some(Reloading{..})` before exec; `None` otherwise (today).
    Bye {
        #[serde(default)]
        reason: Option<ByeReason>,
    },
    /// Reply to `ClientFrame::AuthState`. Labels, kinds and times only:
    /// never token material, emails, account ids or identity hashes.
    AuthState {
        report: agent_core::auth::AuthStateReport,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "bye", rename_all = "snake_case")]
pub enum ByeReason {
    Reloading { generation: u64, retry_after_ms: u64 },
    Shutdown,
    Detached,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "reason", rename_all = "snake_case")]
pub enum RefuseReason {
    Version {
        daemon_version: u32,
        min: u32,
        max: u32,
    },
    Protocol,
    Busy,
    UnknownSession,
    Config,
    /// `synaps daemon reload` version gate failed (C3).
    ReloadRefused { why: String },
}

/// Auth-status *summary* only — never key material.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Welcome {
    pub protocol_version: u32,
    pub daemon_version: String,
    pub pid: u32,
    pub profile: Option<String>,
    pub sessions: Vec<SessionMeta>,
    pub progressive_tool_disclosure: bool,
    /// Reload counter (starts at 1).
    #[serde(default = "one")]
    pub generation: u64,
    /// The daemon's [`build_id`]: tells two builds of one version apart.
    /// Empty from older daemons (unknown). `daemon_version` stays the plain
    /// semver the reload gate parses.
    #[serde(default)]
    pub daemon_build: String,
}

fn one() -> u64 {
    1
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WireEnvelope {
    pub session_id: SessionId,
    pub seq: u64,
    pub ts: chrono::DateTime<chrono::Utc>,
    pub event: WireSessionEvent,
}

impl From<Envelope> for WireEnvelope {
    fn from(e: Envelope) -> Self {
        Self {
            session_id: e.session_id,
            seq: e.seq,
            ts: e.ts,
            event: e.event.into(),
        }
    }
}

impl From<WireEnvelope> for Envelope {
    fn from(e: WireEnvelope) -> Self {
        Self {
            session_id: e.session_id,
            seq: e.seq,
            ts: e.ts,
            event: e.event.into(),
        }
    }
}

/// `AttachSnapshot` with a serialisable replay.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AttachedWire {
    pub client: ClientId,
    pub meta: SessionMeta,
    pub view: RuntimeView,
    pub conversation: ConversationSnapshot,
    pub streaming: bool,
    pub replay: Vec<WireEnvelope>,
    pub pending_prompts: Vec<PromptRequest>,
    pub clients: Vec<(ClientId, ClientKind)>,
    /// Empty from an older daemon.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub roster: Vec<super::types::ClientInfo>,
    /// Absent from an older daemon, which enforces the owner: `Exclusive`.
    #[serde(default = "ClientInput::wire_default")]
    pub client_input: ClientInput,
    #[serde(default)]
    pub input_owner: Option<ClientId>,
    #[serde(default)]
    pub display_tail: Option<crate::session::display::DisplayTail>,
}

impl AttachedWire {
    pub fn new(client: ClientId, s: AttachSnapshot) -> Self {
        Self {
            client,
            meta: s.meta,
            view: s.view,
            conversation: s.conversation,
            streaming: s.streaming,
            replay: s.replay.into_iter().map(Into::into).collect(),
            pending_prompts: s.pending_prompts,
            clients: s.clients,
            roster: s.roster,
            client_input: s.client_input,
            input_owner: s.input_owner,
            display_tail: s.display_tail,
        }
    }

    pub fn into_snapshot(self) -> (ClientId, AttachSnapshot) {
        (
            self.client,
            AttachSnapshot {
                meta: self.meta,
                view: self.view,
                conversation: self.conversation,
                streaming: self.streaming,
                replay: self.replay.into_iter().map(Into::into).collect(),
                pending_prompts: self.pending_prompts,
                clients: self.clients,
                roster: self.roster,
                client_input: self.client_input,
                input_owner: self.input_owner,
                display_tail: self.display_tail,
            },
        )
    }
}

// ── conversation digest ───────────────────────────────────────────────────────

/// `ConversationSnapshot` minus `api_messages`: what `Conversation` carries
/// on the wire. `messages_hash` is FNV-1a over each message's JSON so a
/// client can tell whether its local mirror is current.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ConversationDigest {
    #[serde(default)]
    pub header: SessionHeader,
    pub messages_len: usize,
    pub messages_hash: u64,
    pub tokens: ConversationTokens,
    pub cost: f64,
    #[serde(default)]
    pub keepwarm_cost: f64,
    #[serde(default)]
    pub keepwarm_tokens: ConversationTokens,
    pub abort_context: Option<String>,
    pub queued_message: Option<String>,
    pub pending_events_len: usize,
    pub consecutive_auto_turns: u32,
}

/// FNV-1a 64 over the canonical JSON of every message (stable across
/// binaries, unlike `DefaultHasher`).
pub fn messages_hash(messages: &[crate::SharedMessage]) -> u64 {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    let mut feed = |b: u8| {
        h ^= b as u64;
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    };
    for m in messages {
        if let Ok(bytes) = serde_json::to_vec(&**m) {
            for b in bytes {
                feed(b);
            }
        }
        feed(0x1f);
    }
    h
}

impl ConversationDigest {
    pub fn of(s: &ConversationSnapshot) -> Self {
        Self {
            header: s.header.clone(),
            messages_len: s.api_messages.len(),
            messages_hash: s.messages_hash_memo.hash(&s.api_messages),
            tokens: s.tokens.clone(),
            cost: s.cost,
            keepwarm_cost: s.keepwarm_cost,
            keepwarm_tokens: s.keepwarm_tokens.clone(),
            abort_context: s.abort_context.clone(),
            queued_message: s.queued_message.clone(),
            pending_events_len: s.pending_events_len,
            consecutive_auto_turns: s.consecutive_auto_turns,
        }
    }

    pub fn matches(&self, messages: &[crate::SharedMessage]) -> bool {
        self.messages_len == messages.len() && self.messages_hash == messages_hash(messages)
    }

    /// Rebuild a snapshot around `messages` (the caller's mirror).
    pub fn into_snapshot(self, api_messages: Vec<crate::SharedMessage>) -> ConversationSnapshot {
        ConversationSnapshot {
            header: self.header,
            messages_len: self.messages_len,
            api_messages,
            tokens: self.tokens,
            cost: self.cost,
            keepwarm_cost: self.keepwarm_cost,
            keepwarm_tokens: self.keepwarm_tokens,
            abort_context: self.abort_context,
            queued_message: self.queued_message,
            pending_events_len: self.pending_events_len,
            consecutive_auto_turns: self.consecutive_auto_turns,
            messages_hash_memo: Default::default(),
        }
    }
}

// ── mirror of SessionEventWire (lossless except Conversation → digest) ────────

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "ev", rename_all = "snake_case")]
#[allow(clippy::large_enum_variant)] // mirrors SessionEventWire by design
pub enum WireSessionEvent {
    Stream { event: WireStreamEvent },
    TurnStarted {
        turn_baseline: usize,
        trigger: TurnTrigger,
        #[serde(default)]
        user_text: Option<String>,
    },
    /// Digest only — see module docs.
    Conversation { digest: ConversationDigest },
    Prompt { request: PromptRequest },
    PromptResolved { prompt_id: u64 },
    External { event: crate::events::types::Event },
    AutoTurnCapReached { cap: u32 },
    Idle,
    Steered { text: String, delivered: bool },
    Dequeued { text: String },
    SystemNotice { text: String },
    LoaderProgress { event: WireLoaderEvent },
    ExtensionNotification { extension_id: String, method: String, params: serde_json::Value },
    SettingChanged { applied: SettingApplied },
    QueryResult { id: u64, value: serde_json::Value },
    Attached { client: ClientId, snapshot: AttachedWire },
    ClientJoined {
        client: ClientId,
        kind: ClientKind,
        /// Empty from an older daemon (the client falls back to the kind).
        #[serde(default)]
        name: String,
        /// Absent from an older daemon (read as `Mirror`).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        mode: Option<AttachMode>,
    },
    ClientLeft { client: ClientId },
    Ended { reason: EndReason },
    Aborted { context_saved: bool },
    Cleared { session_id: String },
    CompactionStarted { source: String, disclosure: String },
    CompactionApplied {
        previous_session_id: String,
        session_id: String,
        chains_advanced: Vec<String>,
        queued_restored: Option<String>,
        msg_count: usize,
    },
    CompactionFailed { message: String, panicked: bool },
    CompactionCancelled,
    SubagentRows { rows: Vec<crate::runtime::subagent::SubagentDisplayRow> },
    Resumed {
        id: u64,
        old_id: String,
        new_id: String,
        via: Option<String>,
        clamp_notice: Option<String>,
    },
    InputOwnerChanged { from: Option<ClientId>, to: Option<ClientId>, reason: OwnerChangeReason },
    Refused { client: ClientId, command: String, reason: String },
    AttachRefused { message: String },
    Lifecycle { lifecycle: SessionLifecycle },
    Reloading { generation: u64, retry_after_ms: u64 },
    /// (E-P0) Driver armed.
    DriverArmed {
        plugin_id: String,
        run_id: String,
        models: Vec<crate::extensions::session_driver::Selection>,
        selection: crate::extensions::session_driver::Selection,
        #[serde(default)]
        deadline_ms: Option<u64>,
        #[serde(default)]
        notice: String,
    },
    /// (E-P0) Driver revoked.
    DriverRevoked {
        reason: String,
        #[serde(default)]
        undelivered_steering: Vec<String>,
    },
    /// (E-P0) Driver turn outcome.
    DriverTurnOutcome {
        outcome: crate::extensions::session_driver::Outcome,
        selection: crate::extensions::session_driver::Selection,
        #[serde(default)]
        feedback: Option<String>,
    },
    /// (E-P7) Spend ceiling breached; turn cancelled and driver revoked.
    CostCapReached {
        scope: String,
        cost: f64,
        cap: f64,
    },
    /// Forward-compat: an additive variant from a newer daemon.
    #[serde(other)]
    Unknown,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum WireStreamEvent {
    Llm(WireLlmEvent),
    Session(WireSessionStreamEvent),
    Agent(WireAgentEvent),
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "llm", rename_all = "snake_case")]
pub enum WireLlmEvent {
    Thinking { text: String },
    Text { text: String },
    ToolUseStart { tool_name: String, tool_id: String },
    ToolUseDelta { tool_id: String, delta: String },
    ToolUse { tool_name: String, tool_id: String, input: serde_json::Value },
    ToolResult { tool_id: String, result: String },
    ToolResultDelta { tool_id: String, delta: String },
    ResponseStart,
    ResponseReset,
    /// Opt-in (`ClientCapabilities::tool_queue`): this enum has no
    /// `#[serde(other)]`, so an older client fails the WHOLE frame on an
    /// unknown variant — the daemon never sends these two to a client that
    /// did not declare it can decode them (`daemon/conn.rs`).
    ToolQueued { tool_id: String },
    ToolStarted { tool_id: String },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "session", rename_all = "snake_case")]
pub enum WireSessionStreamEvent {
    MessageHistory { messages: Vec<crate::SharedMessage> },
    Usage {
        input_tokens: u64,
        output_tokens: u64,
        cache_read_input_tokens: u64,
        cache_creation_input_tokens: u64,
        cache_creation_5m: Option<u64>,
        cache_creation_1h: Option<u64>,
        model: Option<String>,
    },
    Done,
    Error { message: String, outcome: TurnOutcome },
    /// `retry`: a `SessionEvent::Retry`. Omitted when false, so a plain
    /// notice's frame is unchanged and peers that predate the flag read a
    /// retry as a plain notice.
    Notice {
        text: String,
        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
        retry: bool,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "agent", rename_all = "snake_case")]
pub enum WireAgentEvent {
    SubagentStart { subagent_id: u64, agent_name: String, task_preview: String },
    SubagentUpdate { subagent_id: u64, agent_name: String, status: String },
    SubagentDone { subagent_id: u64, agent_name: String, result_preview: String, duration_secs: f64 },
    SteeringDelivered { message: String },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WireLoadFailure {
    pub plugin: String,
    pub manifest_path: Option<PathBuf>,
    pub reason: String,
    pub hint: String,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "loader", rename_all = "snake_case")]
pub enum WireLoaderEvent {
    Started,
    Loaded { plugin: String, loaded: usize, failed: usize },
    Failed { failure: WireLoadFailure, loaded: usize, failed: usize },
    Finished { loaded: Vec<String>, failed: Vec<WireLoadFailure> },
}

type LoadFailure = crate::extensions::manager::ExtensionLoadFailure;
type LoaderEvent = crate::extensions::loader::ExtensionLoaderEvent;

impl From<LoadFailure> for WireLoadFailure {
    fn from(f: LoadFailure) -> Self {
        Self { plugin: f.plugin, manifest_path: f.manifest_path, reason: f.reason, hint: f.hint }
    }
}
impl From<WireLoadFailure> for LoadFailure {
    fn from(f: WireLoadFailure) -> Self {
        Self { plugin: f.plugin, manifest_path: f.manifest_path, reason: f.reason, hint: f.hint }
    }
}
impl From<LoaderEvent> for WireLoaderEvent {
    fn from(e: LoaderEvent) -> Self {
        match e {
            LoaderEvent::Started => Self::Started,
            LoaderEvent::Loaded { plugin, loaded, failed } => Self::Loaded { plugin, loaded, failed },
            LoaderEvent::Failed { failure, loaded, failed } => {
                Self::Failed { failure: failure.into(), loaded, failed }
            }
            LoaderEvent::Finished { loaded, failed } => Self::Finished {
                loaded,
                failed: failed.into_iter().map(Into::into).collect(),
            },
        }
    }
}
impl From<WireLoaderEvent> for LoaderEvent {
    fn from(e: WireLoaderEvent) -> Self {
        match e {
            WireLoaderEvent::Started => Self::Started,
            WireLoaderEvent::Loaded { plugin, loaded, failed } => Self::Loaded { plugin, loaded, failed },
            WireLoaderEvent::Failed { failure, loaded, failed } => {
                Self::Failed { failure: failure.into(), loaded, failed }
            }
            WireLoaderEvent::Finished { loaded, failed } => Self::Finished {
                loaded,
                failed: failed.into_iter().map(Into::into).collect(),
            },
        }
    }
}

impl From<StreamEvent> for WireStreamEvent {
    fn from(e: StreamEvent) -> Self {
        match e {
            StreamEvent::Llm(l) => Self::Llm(match l {
                LlmEvent::Thinking(text) => WireLlmEvent::Thinking { text },
                LlmEvent::Text(text) => WireLlmEvent::Text { text },
                LlmEvent::ToolUseStart { tool_name, tool_id } => WireLlmEvent::ToolUseStart { tool_name, tool_id },
                LlmEvent::ToolUseDelta { tool_id, delta } => WireLlmEvent::ToolUseDelta { tool_id, delta },
                LlmEvent::ToolUse { tool_name, tool_id, input } => WireLlmEvent::ToolUse { tool_name, tool_id, input },
                LlmEvent::ToolResult { tool_id, result } => WireLlmEvent::ToolResult { tool_id, result },
                LlmEvent::ToolResultDelta { tool_id, delta } => WireLlmEvent::ToolResultDelta { tool_id, delta },
                LlmEvent::ResponseStart => WireLlmEvent::ResponseStart,
                LlmEvent::ResponseReset => WireLlmEvent::ResponseReset,
                LlmEvent::ToolQueued { tool_id } => WireLlmEvent::ToolQueued { tool_id },
                LlmEvent::ToolStarted { tool_id } => WireLlmEvent::ToolStarted { tool_id },
            }),
            StreamEvent::Session(s) => Self::Session(match s {
                SessionEvent::MessageHistory(messages) => WireSessionStreamEvent::MessageHistory { messages },
                SessionEvent::Usage {
                    input_tokens,
                    output_tokens,
                    cache_read_input_tokens,
                    cache_creation_input_tokens,
                    cache_creation_5m,
                    cache_creation_1h,
                    model,
                } => WireSessionStreamEvent::Usage {
                    input_tokens,
                    output_tokens,
                    cache_read_input_tokens,
                    cache_creation_input_tokens,
                    cache_creation_5m,
                    cache_creation_1h,
                    model,
                },
                SessionEvent::Done => WireSessionStreamEvent::Done,
                SessionEvent::Error(TurnError { message, outcome }) => WireSessionStreamEvent::Error { message, outcome },
                SessionEvent::Notice(text) => WireSessionStreamEvent::Notice { text, retry: false },
                SessionEvent::Retry(text) => WireSessionStreamEvent::Notice { text, retry: true },
                // merge(112): ContextHeadCheckpoint is actor-internal (Wall 1); never crosses the wire.
                // Drop it silently — the actor handles persistence before forwarding.
                SessionEvent::ContextHeadCheckpoint { .. } => WireSessionStreamEvent::Done,
            }),
            StreamEvent::Agent(a) => Self::Agent(match a {
                AgentEvent::SubagentStart { subagent_id, agent_name, task_preview } => {
                    WireAgentEvent::SubagentStart { subagent_id, agent_name, task_preview }
                }
                AgentEvent::SubagentUpdate { subagent_id, agent_name, status } => {
                    WireAgentEvent::SubagentUpdate { subagent_id, agent_name, status }
                }
                AgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs } => {
                    WireAgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs }
                }
                AgentEvent::SteeringDelivered { message } => WireAgentEvent::SteeringDelivered { message },
            }),
        }
    }
}

impl From<WireStreamEvent> for StreamEvent {
    fn from(e: WireStreamEvent) -> Self {
        match e {
            WireStreamEvent::Llm(l) => Self::Llm(match l {
                WireLlmEvent::Thinking { text } => LlmEvent::Thinking(text),
                WireLlmEvent::Text { text } => LlmEvent::Text(text),
                WireLlmEvent::ToolUseStart { tool_name, tool_id } => LlmEvent::ToolUseStart { tool_name, tool_id },
                WireLlmEvent::ToolUseDelta { tool_id, delta } => LlmEvent::ToolUseDelta { tool_id, delta },
                WireLlmEvent::ToolUse { tool_name, tool_id, input } => LlmEvent::ToolUse { tool_name, tool_id, input },
                WireLlmEvent::ToolResult { tool_id, result } => LlmEvent::ToolResult { tool_id, result },
                WireLlmEvent::ToolResultDelta { tool_id, delta } => LlmEvent::ToolResultDelta { tool_id, delta },
                WireLlmEvent::ResponseStart => LlmEvent::ResponseStart,
                WireLlmEvent::ResponseReset => LlmEvent::ResponseReset,
                WireLlmEvent::ToolQueued { tool_id } => LlmEvent::ToolQueued { tool_id },
                WireLlmEvent::ToolStarted { tool_id } => LlmEvent::ToolStarted { tool_id },
            }),
            WireStreamEvent::Session(s) => Self::Session(match s {
                WireSessionStreamEvent::MessageHistory { messages } => SessionEvent::MessageHistory(messages),
                WireSessionStreamEvent::Usage {
                    input_tokens,
                    output_tokens,
                    cache_read_input_tokens,
                    cache_creation_input_tokens,
                    cache_creation_5m,
                    cache_creation_1h,
                    model,
                } => SessionEvent::Usage {
                    input_tokens,
                    output_tokens,
                    cache_read_input_tokens,
                    cache_creation_input_tokens,
                    cache_creation_5m,
                    cache_creation_1h,
                    model,
                },
                WireSessionStreamEvent::Done => SessionEvent::Done,
                WireSessionStreamEvent::Error { message, outcome } => SessionEvent::Error(TurnError { message, outcome }),
                WireSessionStreamEvent::Notice { text, retry: false } => SessionEvent::Notice(text),
                WireSessionStreamEvent::Notice { text, retry: true } => SessionEvent::Retry(text),
            }),
            WireStreamEvent::Agent(a) => Self::Agent(match a {
                WireAgentEvent::SubagentStart { subagent_id, agent_name, task_preview } => {
                    AgentEvent::SubagentStart { subagent_id, agent_name, task_preview }
                }
                WireAgentEvent::SubagentUpdate { subagent_id, agent_name, status } => {
                    AgentEvent::SubagentUpdate { subagent_id, agent_name, status }
                }
                WireAgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs } => {
                    AgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs }
                }
                WireAgentEvent::SteeringDelivered { message } => AgentEvent::SteeringDelivered { message },
            }),
        }
    }
}

impl From<SessionEventWire> for WireSessionEvent {
    fn from(e: SessionEventWire) -> Self {
        use SessionEventWire as S;
        match e {
            S::Stream(ev) => Self::Stream { event: ev.into() },
            S::TurnStarted { turn_baseline, trigger, user_text } => {
                Self::TurnStarted { turn_baseline, trigger, user_text }
            }
            S::Conversation(snapshot) => Self::Conversation { digest: ConversationDigest::of(&snapshot) },
            S::Prompt(request) => Self::Prompt { request },
            S::PromptResolved { prompt_id } => Self::PromptResolved { prompt_id },
            S::External(event) => Self::External { event },
            S::AutoTurnCapReached { cap } => Self::AutoTurnCapReached { cap },
            S::Idle => Self::Idle,
            S::Steered { text, delivered } => Self::Steered { text, delivered },
            S::Dequeued { text } => Self::Dequeued { text },
            S::SystemNotice(text) => Self::SystemNotice { text },
            S::LoaderProgress(ev) => Self::LoaderProgress { event: ev.into() },
            S::ExtensionNotification { extension_id, method, params } => {
                Self::ExtensionNotification { extension_id, method, params }
            }
            S::SettingChanged(applied) => Self::SettingChanged { applied },
            S::QueryResult { id, value } => Self::QueryResult { id, value },
            S::Attached { client, snapshot } => Self::Attached { client, snapshot: AttachedWire::new(client, snapshot) },
            S::ClientJoined { client, kind, name, mode } => Self::ClientJoined { client, kind, name, mode: Some(mode) },
            S::ClientLeft { client } => Self::ClientLeft { client },
            S::Ended { reason } => Self::Ended { reason },
            S::Aborted { context_saved } => Self::Aborted { context_saved },
            S::Cleared { session_id } => Self::Cleared { session_id },
            S::CompactionStarted { source, disclosure } => Self::CompactionStarted { source, disclosure },
            S::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count } => {
                Self::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count }
            }
            S::CompactionFailed { message, panicked } => Self::CompactionFailed { message, panicked },
            S::CompactionCancelled => Self::CompactionCancelled,
            S::SubagentRows(rows) => Self::SubagentRows { rows },
            S::Resumed { id, old_id, new_id, via, clamp_notice } => {
                Self::Resumed { id, old_id, new_id, via, clamp_notice }
            }
            S::InputOwnerChanged { from, to, reason } => Self::InputOwnerChanged { from, to, reason },
            S::Refused { client, command, reason } => Self::Refused { client, command, reason },
            S::AttachRefused { message } => Self::AttachRefused { message },
            S::Lifecycle(lifecycle) => Self::Lifecycle { lifecycle },
            S::Reloading { generation, retry_after_ms } => Self::Reloading { generation, retry_after_ms },
            S::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice } => {
                Self::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice }
            }
            S::DriverRevoked { reason, undelivered_steering } => {
                Self::DriverRevoked { reason, undelivered_steering }
            }
            S::DriverTurnOutcome { outcome, selection, feedback } => {
                Self::DriverTurnOutcome { outcome, selection, feedback }
            }
            S::CostCapReached { scope, cost, cap } => {
                Self::CostCapReached { scope, cost, cap }
            }
        }
    }
}

impl From<WireSessionEvent> for SessionEventWire {
    fn from(e: WireSessionEvent) -> Self {
        use WireSessionEvent as W;
        match e {
            W::Stream { event } => Self::Stream(event.into()),
            W::TurnStarted { turn_baseline, trigger, user_text } => {
                Self::TurnStarted { turn_baseline, trigger, user_text }
            }
            // Messages are not on the wire; `SocketTransport` fills them from
            // its mirror before handing the envelope up.
            W::Conversation { digest } => Self::Conversation(digest.into_snapshot(Vec::new())),
            W::Prompt { request } => Self::Prompt(request),
            W::PromptResolved { prompt_id } => Self::PromptResolved { prompt_id },
            W::External { event } => Self::External(event),
            W::AutoTurnCapReached { cap } => Self::AutoTurnCapReached { cap },
            W::Idle => Self::Idle,
            W::Steered { text, delivered } => Self::Steered { text, delivered },
            W::Dequeued { text } => Self::Dequeued { text },
            W::SystemNotice { text } => Self::SystemNotice(text),
            W::LoaderProgress { event } => Self::LoaderProgress(event.into()),
            W::ExtensionNotification { extension_id, method, params } => {
                Self::ExtensionNotification { extension_id, method, params }
            }
            W::SettingChanged { applied } => Self::SettingChanged(applied),
            W::QueryResult { id, value } => Self::QueryResult { id, value },
            W::Attached { snapshot, .. } => {
                let (client, snapshot) = snapshot.into_snapshot();
                Self::Attached { client, snapshot }
            }
            W::ClientJoined { client, kind, name, mode } => {
                Self::ClientJoined { client, kind, name, mode: mode.unwrap_or(AttachMode::Mirror) }
            }
            W::ClientLeft { client } => Self::ClientLeft { client },
            W::Ended { reason } => Self::Ended { reason },
            W::Aborted { context_saved } => Self::Aborted { context_saved },
            W::Cleared { session_id } => Self::Cleared { session_id },
            W::CompactionStarted { source, disclosure } => Self::CompactionStarted { source, disclosure },
            W::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count } => {
                Self::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count }
            }
            W::CompactionFailed { message, panicked } => Self::CompactionFailed { message, panicked },
            W::CompactionCancelled => Self::CompactionCancelled,
            W::SubagentRows { rows } => Self::SubagentRows(rows),
            W::Resumed { id, old_id, new_id, via, clamp_notice } => {
                Self::Resumed { id, old_id, new_id, via, clamp_notice }
            }
            W::InputOwnerChanged { from, to, reason } => Self::InputOwnerChanged { from, to, reason },
            W::Refused { client, command, reason } => Self::Refused { client, command, reason },
            W::AttachRefused { message } => Self::AttachRefused { message },
            W::Lifecycle { lifecycle } => Self::Lifecycle(lifecycle),
            W::Reloading { generation, retry_after_ms } => Self::Reloading { generation, retry_after_ms },
            W::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice } => {
                Self::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice }
            }
            W::DriverRevoked { reason, undelivered_steering } => {
                Self::DriverRevoked { reason, undelivered_steering }
            }
            W::DriverTurnOutcome { outcome, selection, feedback } => {
                Self::DriverTurnOutcome { outcome, selection, feedback }
            }
            W::CostCapReached { scope, cost, cap } => {
                Self::CostCapReached { scope, cost, cap }
            }
            W::Unknown => Self::SystemNotice("unknown event from a newer daemon (ignored)".into()),
        }
    }
}

// ── per-client decode gate ────────────────────────────────────────────────────

/// A client that predates `client_input` (no `owner_reason_input`) on a
/// `shared` session. It knows only the owner rule, so the daemon presents
/// the truth in those terms: it owns input (its attach snapshot says so,
/// unless it observes) and never sees `InputOwnerChanged`. Otherwise it
/// would announce "input is owned by …" on attach and "input taken over"
/// whenever another client typed, while it can type all along. Matters
/// most after a `daemon reload` onto a new build, when every open TUI is
/// an older client.
pub fn legacy_shared(caps: ClientCapabilities, client_input: ClientInput) -> bool {
    client_input == ClientInput::Shared && !caps.owner_reason_input
}

/// Whether a client that declared `caps` can decode `event` once it is a
/// wire frame. The enums nested under `Stream` have no `#[serde(other)]`,
/// so an older client does not skip an unknown nested variant: the whole
/// frame fails to decode, `SocketTransport::next_event` treats that as a
/// dead socket and the TUI drops into reconnect — whose `Attached` replay
/// may carry the very same event and fail again. The daemon therefore
/// drops, per connection, every event the client did not opt into
/// (`daemon/conn.rs`, live pump and attach replay alike).
pub fn client_decodes(caps: ClientCapabilities, event: &SessionEventWire) -> bool {
    match event {
        SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::ToolQueued { .. } | LlmEvent::ToolStarted { .. })) => {
            caps.tool_queue
        }
        _ => true,
    }
}

// ── framing helpers ───────────────────────────────────────────────────────────

/// Human-readable form of the cap for error frames.
pub fn frame_limit_msg() -> String {
    format!("frame exceeds {} MiB limit", DAEMON_MAX_FRAME_BYTES / (1024 * 1024))
}

/// Encode one frame as a single line (no embedded newline: serde_json
/// escapes them). Refuses to produce a line over [`DAEMON_MAX_FRAME_BYTES`].
pub fn encode_line<T: Serialize>(frame: &T) -> Result<String, String> {
    let mut s = serde_json::to_string(frame).map_err(|e| e.to_string())?;
    if s.len() + 1 > DAEMON_MAX_FRAME_BYTES {
        return Err(frame_limit_msg());
    }
    s.push('\n');
    Ok(s)
}

/// Decode one line, enforcing [`DAEMON_MAX_FRAME_BYTES`].
pub fn decode_line<'a, T: Deserialize<'a>>(line: &'a str) -> Result<T, String> {
    if line.len() > DAEMON_MAX_FRAME_BYTES {
        return Err(frame_limit_msg());
    }
    serde_json::from_str(line).map_err(|e| e.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::BudgetDimension;
    use std::sync::Arc;

    fn view() -> RuntimeView {
        crate::session::handle::echo::view_for()
    }

    /// An `AuthState` reply with every nested type present.
    fn auth_state_report() -> agent_core::auth::AuthStateReport {
        use agent_core::auth::auth_state::{AccountState, CooldownReason, CooldownState, IdentityKind, ObservedOutcome, SelectionState};
        use agent_core::auth::quota_policy::{
            AccountFailure, AccountFailureKind, ObservationFailure, SelectionFailure, SelectionFailureKind,
        };
        let usage_429 = ObservationFailure { kind: AccountFailureKind::UsageRateLimited, http_status: Some(429), retry_after_s: Some(120) };
        agent_core::auth::AuthStateReport {
            schema_version: agent_core::auth::AUTH_STATE_SCHEMA_VERSION,
            live: true,
            broker_active: true,
            generated_at_ms: 1_800_000_000_000,
            store: "/scratch/auth.json".into(),
            accounts: vec![AccountState {
                provider: "anthropic".into(),
                label: "a".into(),
                identity: IdentityKind::Stable,
                access_expires_ms: Some(1_800_000_600_000),
                cooldown: Some(CooldownState { until_ms: 1_800_003_600_000, reason: CooldownReason::AnthropicQuotaExhausted }),
                last_usage: Some(ObservedOutcome { at_ms: 1_799_999_990_000, failure: Some(usage_429.clone()) }),
                last_usage_ok_ms: Some(1_799_999_000_000),
                last_refresh: Some(ObservedOutcome { at_ms: 1_799_999_500_000, failure: None }),
            }],
            selections: vec![SelectionState {
                provider: "anthropic".into(),
                model: Some("claude-x".into()),
                at_ms: 1_799_999_990_000,
                chosen: None,
                failure: Some(SelectionFailure {
                    kind: SelectionFailureKind::UsageReadFailed,
                    accounts: vec![AccountFailure {
                        account: "a".into(),
                        kind: AccountFailureKind::UsageRateLimited,
                        http_status: Some(429),
                        retry_after_s: Some(120),
                        until_ms: None,
                        cooling_down: false,
                    }],
                    earliest_reset_ms: None,
                    usage_read_retry_ms: None,
                }),
                vend_failure: None,
            }],
            cleared: vec!["anthropic:b".into()],
        }
    }

    fn meta() -> SessionMeta {
        crate::session::handle::echo::meta_for(&SessionId::from("wire-1"))
    }

    fn prompt() -> PromptRequest {
        PromptRequest {
            id: 3,
            kind: PromptKind::Secret,
            title: "sudo".into(),
            prompt: "password".into(),
            raised_at: chrono::Utc::now(),
            confirm_scope: None,
        }
    }

    fn conv() -> ConversationSnapshot {
        ConversationSnapshot {
            header: SessionHeader {
                id: "wire-1".into(),
                title: "t".into(),
                name: None,
                model: "m".into(),
                thinking_level: "off".into(),
                system_prompt: Some("sys".into()),
                created_at: chrono::Utc::now(),
                updated_at: chrono::Utc::now(),
                parent_session: Some("p".into()),
            },
            api_messages: vec![Arc::new(serde_json::json!({"role":"user","content":"hi"}))],
            messages_len: 1,
            tokens: ConversationTokens { input: 1, output: 2, cache_read: 3, cache_creation: 4 },
            cost: 0.5,
            keepwarm_cost: 0.125,
            keepwarm_tokens: ConversationTokens { input: 0, output: 0, cache_read: 9, cache_creation: 0 },
            abort_context: Some("ctx".into()),
            queued_message: None,
            pending_events_len: 2,
            consecutive_auto_turns: 1,
            messages_hash_memo: Default::default(),
        }
    }

    /// One snapshot fanned out to N socket clients is hashed ONCE (the memo
    /// is shared by its clones), and a snapshot whose messages were replaced
    /// afterwards is re-hashed, never served a stale hash.
    #[test]
    fn conversation_digest_hash_is_computed_once_and_never_stale() {
        let a = conv();
        let b = a.clone();
        assert_eq!(ConversationDigest::of(&a), ConversationDigest::of(&b));
        assert!(a.messages_hash_memo.0.lock().unwrap().is_some(), "memo shared");
        assert_eq!(
            ConversationDigest::of(&b).messages_hash,
            messages_hash(&a.api_messages)
        );
        let mut c = a.clone();
        c.api_messages = Vec::new();
        assert_eq!(ConversationDigest::of(&c).messages_hash, messages_hash(&[]));
        let mut d = a.clone();
        d.api_messages.push(Arc::new(serde_json::json!({"role":"assistant","content":"yo"})));
        assert_eq!(
            ConversationDigest::of(&d).messages_hash,
            messages_hash(&d.api_messages)
        );
    }

    fn env(event: SessionEventWire) -> Envelope {
        Envelope { session_id: SessionId::from("wire-1"), seq: 9, ts: chrono::Utc::now(), event }
    }

    fn stream_fixtures() -> Vec<StreamEvent> {
        vec![
            StreamEvent::Llm(LlmEvent::Thinking("t".into())),
            StreamEvent::Llm(LlmEvent::Text("x\ny".into())),
            StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name: "bash".into(), tool_id: "id1".into() }),
            StreamEvent::Llm(LlmEvent::ToolUseDelta { tool_id: "id1".into(), delta: "{\"c".into() }),
            StreamEvent::Llm(LlmEvent::ToolUse {
                tool_name: "bash".into(),
                tool_id: "id1".into(),
                input: serde_json::json!({"command":"pwd"}),
            }),
            StreamEvent::Llm(LlmEvent::ToolResult { tool_id: "id1".into(), result: "/tmp".into() }),
            StreamEvent::Llm(LlmEvent::ToolResultDelta { tool_id: "id1".into(), delta: "/t".into() }),
            StreamEvent::Llm(LlmEvent::ToolQueued { tool_id: "id2".into() }),
            StreamEvent::Llm(LlmEvent::ToolStarted { tool_id: "id2".into() }),
            StreamEvent::Session(SessionEvent::MessageHistory(vec![Arc::new(serde_json::json!({"role":"user","content":"a"}))])),
            StreamEvent::Session(SessionEvent::Usage {
                input_tokens: 1,
                output_tokens: 2,
                cache_read_input_tokens: 3,
                cache_creation_input_tokens: 4,
                cache_creation_5m: Some(5),
                cache_creation_1h: None,
                model: Some("m".into()),
            }),
            StreamEvent::Session(SessionEvent::Done),
            StreamEvent::Session(SessionEvent::Error(TurnError::provider("boom", "auth_error", "turn-1-0"))),
            StreamEvent::Session(SessionEvent::Error(TurnError::budget(BudgetDimension::ToolCalls))),
            StreamEvent::Session(SessionEvent::Error(TurnError { message: "c".into(), outcome: TurnOutcome::Canceled })),
            StreamEvent::Session(SessionEvent::Notice("n".into())),
            StreamEvent::Session(SessionEvent::Retry("r".into())),
            StreamEvent::Agent(AgentEvent::SubagentStart { subagent_id: 1, agent_name: "a".into(), task_preview: "p".into() }),
            StreamEvent::Agent(AgentEvent::SubagentUpdate { subagent_id: 1, agent_name: "a".into(), status: "s".into() }),
            StreamEvent::Agent(AgentEvent::SubagentDone {
                subagent_id: 1,
                agent_name: "a".into(),
                result_preview: "r".into(),
                duration_secs: 1.5,
            }),
            StreamEvent::Agent(AgentEvent::SteeringDelivered { message: "m".into() }),
        ]
    }

    /// 04-tui I4: a retry rides on the existing `notice` frame with a flag,
    /// so a peer that predates it decodes it as a plain notice (it never
    /// becomes a malformed frame), and a plain notice's bytes are unchanged.
    #[test]
    fn a_retry_is_a_flagged_notice_on_the_wire() {
        let retry = StreamEvent::Session(SessionEvent::Retry("⚠ Rate limited — retrying (1/5)".into()));
        let json = serde_json::to_value(WireStreamEvent::from(retry)).unwrap();
        assert_eq!(json["session"], "notice", "{json}");
        assert_eq!(json["retry"], true, "{json}");
        let back: WireStreamEvent = serde_json::from_value(json).unwrap();
        assert!(
            matches!(StreamEvent::from(back), StreamEvent::Session(SessionEvent::Retry(t)) if t.starts_with("⚠ Rate limited")),
            "a retry stays a retry"
        );
        let plain = StreamEvent::Session(SessionEvent::Notice("n".into()));
        let json = serde_json::to_value(WireStreamEvent::from(plain)).unwrap();
        assert!(json.get("retry").is_none(), "a plain notice is unchanged: {json}");
        // An older daemon's notice has no flag: a plain notice.
        let old: WireStreamEvent =
            serde_json::from_value(serde_json::json!({"kind": "session", "session": "notice", "text": "x"})).unwrap();
        assert!(matches!(StreamEvent::from(old), StreamEvent::Session(SessionEvent::Notice(t)) if t == "x"));
    }

    fn fixtures() -> Vec<SessionEventWire> {
        use SessionEventWire as S;
        let loader = crate::extensions::loader::ExtensionLoaderEvent::Finished {
            loaded: vec!["p".into()],
            failed: vec![LoadFailure {
                plugin: "bad".into(),
                manifest_path: Some(PathBuf::from("/x/manifest.json")),
                reason: "oops".into(),
                hint: "fix".into(),
            }],
        };
        let mut v: Vec<S> = stream_fixtures().into_iter().map(S::Stream).collect();
        v.extend([
            S::TurnStarted { turn_baseline: 4, trigger: TurnTrigger::EventAuto, user_text: None },
            S::TurnStarted { turn_baseline: 5, trigger: TurnTrigger::QueuedAuto, user_text: Some("q".into()) },
            S::TurnStarted { turn_baseline: 6, trigger: TurnTrigger::DriverAuto, user_text: None },
            S::Conversation(conv()),
            S::Prompt(prompt()),
            S::PromptResolved { prompt_id: 3 },
            S::External(crate::events::types::Event::simple("cli", "hello", None)),
            S::AutoTurnCapReached { cap: 5 },
            S::Idle,
            S::Steered { text: "s".into(), delivered: true },
            S::Dequeued { text: "d".into() },
            S::SystemNotice("sys".into()),
            S::LoaderProgress(loader),
            S::ExtensionNotification { extension_id: "e".into(), method: "widget.upsert".into(), params: serde_json::json!({"a":1}) },
            S::SettingChanged(SettingApplied {
                id: 3,
                setting: "model".into(),
                ok: true,
                message: None,
                view: view(),
                clamp: Some(ReasoningClampWire { from: "ultra".into(), to: "high".into() }),
            }),
            S::QueryResult { id: 1, value: serde_json::json!([1, 2]) },
            S::Attached {
                client: ClientId(2),
                snapshot: AttachSnapshot {
                    meta: meta(),
                    view: view(),
                    conversation: conv(),
                    streaming: true,
                    replay: vec![env(S::Stream(StreamEvent::Llm(LlmEvent::Text("partial".into()))))],
                    pending_prompts: vec![prompt()],
                    clients: vec![(ClientId(1), ClientKind::Tui)],
                    roster: Vec::new(),
                    client_input: ClientInput::Shared,
                    input_owner: Some(ClientId(1)),
                    display_tail: None,
                },
            },
            S::ClientJoined { client: ClientId(1), kind: ClientKind::Attach, name: "discord".into(), mode: AttachMode::Observe },
            S::ClientLeft { client: ClientId(1) },
            S::Ended { reason: EndReason::HostShutdown },
            S::Aborted { context_saved: true },
            S::Cleared { session_id: "new-1".into() },
            S::CompactionStarted { source: "manual".into(), disclosure: "disc".into() },
            S::CompactionApplied {
                previous_session_id: "old".into(),
                session_id: "new".into(),
                chains_advanced: vec!["c1".into()],
                queued_restored: Some("q".into()),
                msg_count: 12,
            },
            S::CompactionFailed { message: "boom".into(), panicked: false },
            S::CompactionCancelled,
            S::SubagentRows(vec![crate::runtime::subagent::SubagentDisplayRow {
                subagent_id: 1,
                agent_name: "a".into(),
                status: crate::runtime::subagent::SubagentStatus::Failed("x".into()),
                cancel_requested: false,
                elapsed_secs: 1.25,
                finished_elapsed: Some(std::time::Duration::from_millis(1250)),
                step: "$ cargo test".into(),
                tools: 3,
                handle_id: "sa_1".into(),
                model: "provider/model-a".into(),
            }]),
            S::Resumed { id: 2, old_id: "o".into(), new_id: "n".into(), via: Some("name".into()), clamp_notice: None },
            S::InputOwnerChanged { from: Some(ClientId(1)), to: Some(ClientId(2)), reason: OwnerChangeReason::Takeover },
            S::Refused { client: ClientId(2), command: "submit".into(), reason: "input owned by client #1".into() },
            S::AttachRefused { message: "parked".into() },
            S::Lifecycle(SessionLifecycle::Parked),
            S::Reloading { generation: 2, retry_after_ms: 500 },
            S::DriverArmed {
                plugin_id: "autonomous".into(),
                run_id: "run-1".into(),
                models: vec![
                    crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
                    crate::extensions::session_driver::Selection { model: "m2".into(), effort: "low".into() },
                ],
                selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
                deadline_ms: Some(60_000),
                notice: "armed".into(),
            },
            S::DriverRevoked { reason: "canceled".into(), undelivered_steering: vec!["steer1".into()] },
            S::DriverTurnOutcome {
                outcome: crate::extensions::session_driver::Outcome::Success,
                selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
                feedback: Some("changed".into()),
            },
        ]);
        v
    }

    fn queued_frame() -> String {
        let ev = env(SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::ToolQueued { tool_id: "toolu_2".into() })));
        encode_line(&DaemonFrame::Event(ev.into())).unwrap()
    }

    /// What an OLDER client does with a new nested `llm` variant: its
    /// `WireLlmEvent` lacks the tag, so serde fails the WHOLE frame (no
    /// `#[serde(other)]` down there) — `SocketTransport::next_event` then
    /// ends the stream as if the socket closed. Simulated the way the
    /// driver-event test does: rename the tag to one this build lacks.
    /// Only the top level (`ev`) has an `Unknown` fallback, and that one
    /// shows the user a notice.
    #[test]
    fn an_unknown_nested_llm_variant_fails_the_whole_frame() {
        let line = queued_frame();
        assert!(line.contains(r#""llm":"tool_queued""#), "{line}");
        let as_old = line.replace(r#""llm":"tool_queued""#, r#""llm":"tool_from_the_future""#);
        let err = decode_line::<DaemonFrame>(as_old.trim_end()).unwrap_err();
        assert!(err.contains("unknown variant `tool_from_the_future`"), "{err}");
        // The same inside an attach replay sinks the whole `Attached`.
        let attached = AttachedWire::new(
            ClientId(1),
            AttachSnapshot {
                meta: meta(),
                view: view(),
                conversation: conv(),
                streaming: true,
                replay: vec![env(SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::ToolQueued { tool_id: "toolu_2".into() })))],
                pending_prompts: vec![],
                clients: vec![],
                roster: Vec::new(),
                client_input: ClientInput::Shared,
                input_owner: None,
                display_tail: None,
            },
        );
        let line = encode_line(&DaemonFrame::Attached(attached)).unwrap();
        let as_old = line.replace(r#""llm":"tool_queued""#, r#""llm":"tool_from_the_future""#);
        assert!(decode_line::<DaemonFrame>(as_old.trim_end()).is_err());
        // Whereas an unknown TOP-level event decodes, as a user-visible notice.
        let top_frame = queued_frame().replace(r#""ev":"stream""#, r#""ev":"zz_new""#);
        match decode_line::<DaemonFrame>(top_frame.trim_end()).unwrap() {
            DaemonFrame::Event(w) => match Envelope::from(w).event {
                SessionEventWire::SystemNotice(t) => assert!(t.contains("newer daemon"), "{t}"),
                other => panic!("{other:?}"),
            },
            other => panic!("{other:?}"),
        }
    }

    /// `client_input` on the wire: a current daemon sends it; an `Attached`
    /// from an older daemon (no field) means `Exclusive`, the owner rule it
    /// enforces. Only a client without `owner_reason_input` on a shared
    /// session is presented the owner rule.
    #[test]
    fn client_input_absent_means_exclusive_and_legacy_clients_are_presented_owners() {
        let snap = |client_input| AttachSnapshot {
            meta: meta(),
            view: view(),
            conversation: conv(),
            streaming: false,
            replay: vec![],
            pending_prompts: vec![],
            clients: vec![],
            roster: Vec::new(),
            client_input,
            input_owner: Some(ClientId(1)),
            display_tail: None,
        };
        let line = encode_line(&DaemonFrame::Attached(AttachedWire::new(ClientId(2), snap(ClientInput::Shared)))).unwrap();
        assert!(line.contains(r#""client_input":"shared""#), "{line}");
        let decoded = |line: &str| match decode_line::<DaemonFrame>(line.trim_end()).unwrap() {
            DaemonFrame::Attached(a) => a.into_snapshot().1,
            other => panic!("{other:?}"),
        };
        assert_eq!(decoded(&line).client_input, ClientInput::Shared);
        let old = line.replace(r#""client_input":"shared","#, "");
        assert!(!old.contains("client_input"), "{old}");
        let s = decoded(&old);
        assert_eq!(s.client_input, ClientInput::Exclusive);
        assert!(s.input_owned_elsewhere(ClientId(2)).is_some(), "an older daemon's non-owner is read-only");

        let (old_caps, new_caps) = (ClientCapabilities::default(), ClientCapabilities::current());
        assert!(legacy_shared(old_caps, ClientInput::Shared));
        assert!(!legacy_shared(new_caps, ClientInput::Shared));
        assert!(!legacy_shared(old_caps, ClientInput::Exclusive), "exclusive is the rule it knows");
    }

    /// The gate: a client that did not declare `tool_queue` (every client
    /// that predates it) is never sent the queue signals; everything else
    /// it decodes exactly as before. A current client gets them.
    #[test]
    fn queue_signals_reach_only_clients_that_opted_in() {
        let old = ClientCapabilities::default();
        let new = ClientCapabilities::current();
        for ev in fixtures() {
            let queue = matches!(
                ev,
                SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::ToolQueued { .. } | LlmEvent::ToolStarted { .. }))
            );
            assert_eq!(client_decodes(old, &ev), !queue, "old client: {ev:?}");
            assert!(client_decodes(new, &ev), "new client: {ev:?}");
        }
    }

    /// Hello compat both ways: an older client's meta has no capabilities
    /// (decodes as none), a current one declares `tool_queue`, and an older
    /// daemon's `ClientMeta` (no such field) still decodes the new frame.
    #[test]
    fn client_capabilities_default_off_and_old_daemons_ignore_them() {
        let legacy = r#"{"kind":"tui","terminal":null,"instance":"i"}"#;
        let m: ClientMeta = serde_json::from_str(legacy).unwrap();
        assert_eq!(m.capabilities, ClientCapabilities::default());
        assert!(!m.capabilities.tool_queue);

        let h = Hello::new(ClientKind::Tui);
        assert!(h.client.capabilities.tool_queue);
        let line = encode_line(&ClientFrame::Hello(h)).unwrap();
        assert!(line.contains(r#""capabilities":{"tool_queue":true,"owner_reason_input":true}"#), "{line}");
        match decode_line::<ClientFrame>(line.trim_end()).unwrap() {
            ClientFrame::Hello(h) => assert!(h.client.capabilities.tool_queue),
            other => panic!("{other:?}"),
        }

        // The ClientMeta an older daemon decodes (as of da1eb372).
        #[derive(Deserialize)]
        #[allow(dead_code)]
        struct OldClientMeta {
            kind: ClientKind,
            terminal: Option<String>,
            instance: String,
            #[serde(default)]
            history: HistoryMode,
            #[serde(default)]
            tail_items: usize,
        }
        let json = serde_json::to_string(&ClientMeta::new(ClientKind::Tui)).unwrap();
        let old: OldClientMeta = serde_json::from_str(&json).unwrap();
        assert_eq!(old.kind, ClientKind::Tui);
    }

    /// Client names, both directions: an older daemon's `ClientJoined` (no
    /// name, no mode) decodes as the kind's name in `Mirror`; a `Hello`
    /// without a name sends no `name` key (older daemons see the old
    /// bytes), with one it round-trips.
    #[test]
    fn client_names_are_additive_on_the_wire() {
        let old = r#"{"ev":"client_joined","client":4,"kind":"attach"}"#;
        let ev: SessionEventWire = serde_json::from_str::<WireSessionEvent>(old).unwrap().into();
        let mut roster = Roster::default();
        roster.apply(&ev);
        let c = roster.get(ClientId(4)).unwrap();
        assert_eq!((c.name.as_str(), c.mode), ("attach", AttachMode::Mirror));

        let unnamed = encode_line(&ClientFrame::Hello(Hello::new(ClientKind::Tui))).unwrap();
        assert!(!unnamed.contains(r#""name""#), "{unnamed}");
        let named = Hello::new(ClientKind::Tui).with_name(Some("discord".into()));
        let line = encode_line(&ClientFrame::Hello(named)).unwrap();
        match decode_line::<ClientFrame>(line.trim_end()).unwrap() {
            ClientFrame::Hello(h) => assert_eq!(h.client.name.as_deref(), Some("discord")),
            other => panic!("{other:?}"),
        }
    }

    /// The watch query's opt-in: an older client's query omits it (off), an
    /// off flag is not serialized (an older daemon sees the old bytes).
    #[test]
    fn the_watch_queries_tool_queue_flag_defaults_off() {
        let old: SessionQuery =
            serde_json::from_str(r#"{"query":"subagent_watch","handle_id":"sa_1","since":4}"#).unwrap();
        assert!(matches!(old, SessionQuery::SubagentWatch { tool_queue: false, since: 4, .. }));
        let off = SessionQuery::SubagentWatch { handle_id: "sa_1".into(), since: 4, tool_queue: false };
        assert!(!serde_json::to_string(&off).unwrap().contains("tool_queue"));
        let on = SessionQuery::SubagentWatch { handle_id: "sa_1".into(), since: 4, tool_queue: true };
        let back: SessionQuery = serde_json::from_str(&serde_json::to_string(&on).unwrap()).unwrap();
        assert!(matches!(back, SessionQuery::SubagentWatch { tool_queue: true, .. }));
    }

    #[test]
    fn wire_roundtrip_every_variant() {
        let all = fixtures();
        // 36 non-Stream variants (TurnStarted thrice = 3; +3 driver events) + 21 StreamEvent leaves.
        assert_eq!(all.len(), 36 + 21);
        for ev in all {
            let is_conv = matches!(ev, SessionEventWire::Conversation(_));
            let e = env(ev);
            let before = format!("{e:?}");
            let wire: WireEnvelope = e.into();
            let line = encode_line(&wire).unwrap();
            assert_eq!(line.matches('\n').count(), 1, "one line per frame");
            let back: WireEnvelope = decode_line(line.trim_end()).unwrap();
            let after = format!("{:?}", Envelope::from(back));
            if is_conv {
                // Digest on the wire: everything but the messages survives.
                assert!(!line.contains("api_messages"), "{line}");
                assert!(after.contains("api_messages: []"), "{after}");
                assert_eq!(before.replace("api_messages: [Object {\"content\": String(\"hi\"), \"role\": String(\"user\")}]", "api_messages: []"), after);
            } else {
                assert_eq!(before, after);
            }
        }
    }

    #[test]
    fn hello_history_fields_default_and_roundtrip() {
        // Old peers omit the fields: Full / 120.
        let legacy = r#"{"kind":"tui","terminal":null,"instance":"i"}"#;
        let m: ClientMeta = serde_json::from_str(legacy).unwrap();
        assert_eq!(m.history, HistoryMode::Full);
        assert_eq!(m.tail_items, DEFAULT_TAIL_ITEMS);
        let h = Hello::new(ClientKind::Tui).with_history(HistoryMode::Digest).with_tail_items(9);
        let line = encode_line(&ClientFrame::Hello(h)).unwrap();
        assert!(line.contains("\"history\":\"digest\""), "{line}");
        assert!(line.contains("\"tail_items\":9"), "{line}");
        match decode_line::<ClientFrame>(line.trim_end()).unwrap() {
            ClientFrame::Hello(h) => {
                assert_eq!(h.client.history, HistoryMode::Digest);
                assert_eq!(h.client.tail_items, 9);
            }
            other => panic!("{other:?}"),
        }
        // Digest snapshots keep messages_len even with api_messages empty.
        let d = ConversationDigest::of(&conv());
        let snap = d.into_snapshot(Vec::new());
        assert_eq!(snap.messages_len, 1);
        assert!(snap.api_messages.is_empty());
        let legacy: ConversationSnapshot = serde_json::from_str(r#"{"api_messages":[],"tokens":{"input":0,"output":0,"cache_read":0,"cache_creation":0},"cost":0.0,"abort_context":null,"queued_message":null,"pending_events_len":0,"consecutive_auto_turns":0}"#).unwrap();
        assert_eq!(legacy.messages_len, 0);
    }

    #[test]
    fn conversation_digest_roundtrips_and_detects_drift() {
        let c = conv();
        let d = ConversationDigest::of(&c);
        assert_eq!(d.messages_len, 1);
        assert!(d.matches(&c.api_messages));
        assert!(!d.matches(&[]));
        let mut other = c.api_messages.clone();
        other.push(Arc::new(serde_json::json!({"role":"assistant","content":"x"})));
        assert!(!d.matches(&other));
        let back = d.clone().into_snapshot(c.api_messages.clone());
        assert_eq!(ConversationDigest::of(&back), d);
        // stable across runs/binaries: FNV-1a offset basis for the empty list
        assert_eq!(messages_hash(&[]), 0xcbf2_9ce4_8422_2325);
        assert_ne!(messages_hash(&c.api_messages), messages_hash(&[]));
    }

    #[test]
    fn unknown_event_variant_tolerated() {
        let v: WireSessionEvent = serde_json::from_str(r#"{"ev":"from_the_future","x":1}"#).unwrap();
        assert!(matches!(v, WireSessionEvent::Unknown));
        assert!(matches!(SessionEventWire::from(v), SessionEventWire::SystemNotice(_)));
    }

    /// An older daemon cannot decode a newer client's command; serde names
    /// the unknown tag in backticks. The TUI keys on exactly that
    /// (`tui::watch::absorb_notice`, "malformed frame: unknown variant
    /// `subagent_watch`") to tell the user to reload the daemon.
    #[test]
    fn an_unknown_command_or_query_names_its_tag_in_the_decode_error() {
        let frames = [
            (
                ClientFrame::Cmd {
                    session_id: "s".into(),
                    cmd: SessionCommand::Query {
                        id: 1,
                        query: SessionQuery::SubagentWatch { handle_id: "sa_1".into(), since: 0, tool_queue: false },
                    },
                },
                "subagent_watch",
            ),
            (
                ClientFrame::Cmd {
                    session_id: "s".into(),
                    cmd: SessionCommand::SubagentSteer { handle_id: "sa_1".into(), text: "x".into() },
                },
                "subagent_steer",
            ),
        ];
        for (frame, tag) in frames {
            let json = serde_json::to_string(&frame).unwrap();
            assert!(decode_line::<ClientFrame>(&json).is_ok());
            // What an older daemon sees: a tag it does not know.
            let old = json.replace(&format!("\"{tag}\""), "\"zz_from_the_future\"");
            assert_ne!(old, json, "{tag} is serialized as a quoted tag");
            let err = decode_line::<ClientFrame>(&old).unwrap_err();
            assert!(err.contains("unknown variant `zz_from_the_future`"), "{err}");
        }
    }

    #[test]
    fn driver_armed_round_trips_and_old_client_skips() {
        // A v3 daemon emits DriverArmed; a v2 client decodes it as Unknown.
        let armed = WireSessionEvent::DriverArmed {
            plugin_id: "autonomous".into(),
            run_id: "run-42".into(),
            models: vec![
                crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
            ],
            selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
            deadline_ms: Some(120_000),
            notice: "armed and ready".into(),
        };
        let json = serde_json::to_string(&armed).unwrap();
        // v3 client: exact round-trip.
        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
        assert!(matches!(&back, WireSessionEvent::DriverArmed { plugin_id, .. } if plugin_id == "autonomous"));
        // Simulated v2 client: rename tag so it hits Unknown.
        let old_json = json.replace(r#""ev":"driver_armed""#, r#""ev":"v2_unknown_event""#);
        let old: WireSessionEvent = serde_json::from_str(&old_json).unwrap();
        assert!(matches!(old, WireSessionEvent::Unknown));
    }

    #[test]
    fn driver_revoked_and_outcome_round_trip() {
        let revoked = WireSessionEvent::DriverRevoked {
            reason: "canceled".into(),
            undelivered_steering: vec!["steer1".into(), "steer2".into()],
        };
        let json = serde_json::to_string(&revoked).unwrap();
        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
        assert!(matches!(&back, WireSessionEvent::DriverRevoked { reason, undelivered_steering }
            if reason == "canceled" && undelivered_steering.len() == 2));

        let outcome = WireSessionEvent::DriverTurnOutcome {
            outcome: crate::extensions::session_driver::Outcome::Success,
            selection: crate::extensions::session_driver::Selection { model: "m".into(), effort: "high".into() },
            feedback: Some("changed".into()),
        };
        let json = serde_json::to_string(&outcome).unwrap();
        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
        assert!(matches!(&back, WireSessionEvent::DriverTurnOutcome { feedback: Some(f), .. } if f == "changed"));
    }

    #[test]
    fn driver_revoked_defaults_tolerate_missing_fields() {
        // Old daemon or minimal frame: undelivered_steering missing → empty vec.
        let minimal = r#"{"ev":"driver_revoked","reason":"timeout"}"#;
        let v: WireSessionEvent = serde_json::from_str(minimal).unwrap();
        assert!(matches!(&v, WireSessionEvent::DriverRevoked { reason, undelivered_steering }
            if reason == "timeout" && undelivered_steering.is_empty()));
    }

    #[test]
    fn client_frames_roundtrip_and_daemon_frames_roundtrip() {
        let frames = vec![
            ClientFrame::Hello(Hello::new(ClientKind::Attach)),
            ClientFrame::Hello(Hello::new(ClientKind::Tui).with_history(HistoryMode::Digest).with_tail_items(7)),
            ClientFrame::Ping,
            ClientFrame::Sessions,
            ClientFrame::Shutdown { force: true },
            ClientFrame::Attach(Attach::Existing { session_id: "s".into(), mode: AttachMode::Mirror }),
            ClientFrame::Attach(Attach::Create { config: SessionConfig::default(), mode: AttachMode::Takeover }),
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Submit { text: "hi".into(), attachments: vec![] } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Answer { prompt_id: 1, value: Some("pw".into()) } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::EngineCommand { id: 4, name: "model".into(), arg: "x".into() } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Query { id: 5, query: SessionQuery::Status } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Set { id: 6, setting: SessionSetting::ReloadPrompt } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Checkpoint { reason: CheckpointReason::Reload } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::KeepWarm { on: true } },
            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::DriverStart {
                plugin: "autonomous".into(), command: "auto".into(), arg: "start -- do stuff".into(),
            }},
            ClientFrame::Reload { now: true, drain_secs: Some(3), exe: Some(PathBuf::from("/bin/synaps")) },
            ClientFrame::Purge,
            ClientFrame::Close { session_id: "s".into(), force: true, attached: true },
            ClientFrame::Close { session_id: "s".into(), force: false, attached: false },
            ClientFrame::Pin { session_id: "s".into(), on: true },
            ClientFrame::Pin { session_id: "s".into(), on: false },
            ClientFrame::AuthState(agent_core::auth::AuthStateQuery::default()),
            ClientFrame::AuthState(agent_core::auth::AuthStateQuery {
                provider: Some("anthropic".into()),
                account: Some("b".into()),
                clear: true,
            }),
            ClientFrame::ReloadExtension { extension: "subconscious".into() },
            ClientFrame::Bye,
        ];
        for f in frames {
            let line = encode_line(&f).unwrap();
            let back: ClientFrame = decode_line(line.trim_end()).unwrap();
            assert_eq!(serde_json::to_string(&back).unwrap(), line.trim_end());
        }
        let frames = vec![
            DaemonFrame::Welcome(Welcome {
                protocol_version: PROTOCOL_VERSION,
                daemon_version: "0.9.0".into(),
                pid: 1,
                profile: None,
                sessions: vec![meta()],
                progressive_tool_disclosure: true,
                generation: 1,
                daemon_build: "abc1234".into(),
            }),
            DaemonFrame::Refused { reason: RefuseReason::Version { daemon_version: 2, min: 2, max: 2 }, message: "no".into() },
            DaemonFrame::Refused { reason: RefuseReason::ReloadRefused { why: "older".into() }, message: "no".into() },
            DaemonFrame::Pong { pid: 1, uptime_s: 2, sessions: 0 },
            DaemonFrame::SessionList { sessions: vec![meta()] },
            DaemonFrame::Attached(AttachedWire::new(
                ClientId(1),
                AttachSnapshot {
                    meta: meta(),
                    view: view(),
                    conversation: conv(),
                    streaming: false,
                    replay: vec![],
                    pending_prompts: vec![],
                    clients: vec![],
                    roster: Vec::new(),
                    client_input: ClientInput::Shared,
                    input_owner: None,
                    display_tail: None,
                },
            )),
            DaemonFrame::Event(env(SessionEventWire::SystemNotice("x".into())).into()),
            DaemonFrame::Error { session_id: None, message: "e".into() },
            DaemonFrame::Closed { session_id: "s".into(), text: "closed".into() },
            DaemonFrame::Pinned { session_id: "s".into(), on: true },
            DaemonFrame::Pinned { session_id: "s".into(), on: false },
            DaemonFrame::ExtensionReloaded {
                report: crate::extensions::manager::ExtensionReloadReport {
                    extension: "subconscious".into(),
                    was_loaded: true,
                    deferred: false,
                    tools: vec!["subconscious:day_note".into()],
                    removed_tools: vec!["subconscious:old".into()],
                    plugin_dir: PathBuf::from("/p/subconscious"),
                },
            },
            DaemonFrame::ExtensionReloadFailed { extension: "x".into(), message: "boom".into(), unloaded: true },
            DaemonFrame::Bye { reason: None },
            DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation: 2, retry_after_ms: 500 }) },
            DaemonFrame::Bye { reason: Some(ByeReason::Shutdown) },
            DaemonFrame::AuthState { report: auth_state_report() },
        ];
        for f in frames {
            let line = encode_line(&f).unwrap();
            let back: DaemonFrame = decode_line(line.trim_end()).unwrap();
            assert_eq!(serde_json::to_string(&back).unwrap(), line.trim_end());
        }
    }

    #[test]
    fn hello_must_be_first_is_a_type_level_fact_and_version_is_exact() {
        let h = Hello::new(ClientKind::Test);
        assert_eq!(h.protocol_version, PROTOCOL_VERSION);
        assert_eq!(PROTOCOL_MIN, PROTOCOL_MAX);
        let json = serde_json::to_string(&ClientFrame::Hello(h)).unwrap();
        assert!(json.starts_with(r#"{"type":"hello""#));
        assert_eq!(PROTOCOL_VERSION, 3);
        // v1 frames (no `reconnect_of`, bare `bye`) still parse — the refusal is by version, not by shape.
        let h: Hello = serde_json::from_str(
            r#"{"protocol_version":1,"client":{"kind":"tui","terminal":null,"instance":"i"},"cwd":"/","client_version":"0"}"#,
        )
        .unwrap();
        assert!(h.reconnect_of.is_none());
        // Legacy Hello without env field → None (backwards compatible).
        assert!(h.env.is_none());
        // Hello::new captures env.
        let h2 = Hello::new(ClientKind::Test);
        assert!(h2.env.is_some(), "Hello::new must capture client env");
        let b: DaemonFrame = serde_json::from_str(r#"{"type":"bye"}"#).unwrap();
        assert!(matches!(b, DaemonFrame::Bye { reason: None }));
    }

    #[test]
    fn close_defaults_force_off_and_older_daemons_reject_it_as_an_unknown_variant() {
        let f: ClientFrame = decode_line(r#"{"type":"close","session_id":"s"}"#).unwrap();
        assert!(matches!(f, ClientFrame::Close { ref session_id, force: false, attached: false } if session_id.as_str() == "s"));
        // A daemon built before `Close` decodes into an enum without it; the
        // client keys "daemon too old" on this exact serde wording.
        #[derive(Debug, Deserialize)]
        #[serde(tag = "type", rename_all = "snake_case")]
        enum OldClientFrame {
            Ping,
            Purge,
        }
        let err = decode_line::<OldClientFrame>(r#"{"type":"close","session_id":"s","force":true}"#).unwrap_err();
        assert!(err.contains("unknown variant `close`"), "{err}");
    }

    #[test]
    fn pin_requires_on_and_older_daemons_reject_it_as_an_unknown_variant() {
        let f: ClientFrame = decode_line(r#"{"type":"pin","session_id":"s","on":false}"#).unwrap();
        assert!(matches!(f, ClientFrame::Pin { ref session_id, on: false } if session_id.as_str() == "s"));
        assert!(decode_line::<ClientFrame>(r#"{"type":"pin","session_id":"s"}"#).is_err());
        #[derive(Debug, Deserialize)]
        #[serde(tag = "type", rename_all = "snake_case")]
        enum OldClientFrame {
            Ping,
            Purge,
        }
        let err = decode_line::<OldClientFrame>(r#"{"type":"pin","session_id":"s","on":true}"#).unwrap_err();
        assert!(err.contains("unknown variant `pin`"), "{err}");
    }

    #[test]
    fn reload_extension_requires_a_name_and_older_daemons_reject_it_as_an_unknown_variant() {
        let f: ClientFrame = decode_line(r#"{"type":"reload_extension","extension":"axel"}"#).unwrap();
        assert!(matches!(f, ClientFrame::ReloadExtension { ref extension } if extension == "axel"));
        assert!(decode_line::<ClientFrame>(r#"{"type":"reload_extension"}"#).is_err());
        // An older daemon's reply-less failure mode: `Connected::reload_extension`
        // keys "daemon too old" on this exact wording.
        #[derive(Debug, Deserialize)]
        #[serde(tag = "type", rename_all = "snake_case")]
        enum OldClientFrame {
            Ping,
            Purge,
        }
        let err = decode_line::<OldClientFrame>(r#"{"type":"reload_extension","extension":"axel"}"#).unwrap_err();
        assert!(err.contains("unknown variant `reload_extension`"), "{err}");
        // `unloaded` is optional on the failure reply.
        let d: DaemonFrame = serde_json::from_str(r#"{"type":"extension_reload_failed","extension":"x","message":"m"}"#).unwrap();
        assert!(matches!(d, DaemonFrame::ExtensionReloadFailed { unloaded: false, .. }));
    }

    #[test]
    fn a_session_meta_without_pinned_decodes_as_not_pinned() {
        let mut v = serde_json::to_value(meta()).unwrap();
        v.as_object_mut().unwrap().remove("pinned");
        let m: SessionMeta = serde_json::from_value(v).unwrap();
        assert!(!m.pinned);
        let mut pinned = meta();
        pinned.pinned = true;
        let back: SessionMeta = serde_json::from_value(serde_json::to_value(&pinned).unwrap()).unwrap();
        assert!(back.pinned);
    }

    #[test]
    fn debug_redacts_answer_and_submit_bodies() {
        let secret = "hunter2-very-secret";
        let f = ClientFrame::Cmd {
            session_id: "s".into(),
            cmd: SessionCommand::Answer { prompt_id: 1, value: Some(secret.into()) },
        };
        let d = format!("{f:?}");
        assert!(!d.contains(secret), "{d}");
        assert!(d.contains("redacted"));
        let f = ClientFrame::Cmd {
            session_id: "s".into(),
            cmd: SessionCommand::Submit { text: "my private prompt".into(), attachments: vec![] },
        };
        let d = format!("{f:?}");
        assert!(!d.contains("private"), "{d}");
        let f = ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Steer { text: "steer me".into() } };
        let d = format!("{f:?}");
        assert!(!d.contains("steer me"));
        // No lengths either: a char count is a side channel.
        assert!(!d.contains("chars"), "{d}");
        assert!(!d.contains('8'), "{d}");
    }

    #[test]
    fn oversize_frame_rejected_both_directions() {
        const { assert!(DAEMON_MAX_FRAME_BYTES > RPC_MAX_FRAME_BYTES) };
        let big = "x".repeat(DAEMON_MAX_FRAME_BYTES + 1);
        assert!(decode_line::<ClientFrame>(&big)
            .unwrap_err()
            .contains("64 MiB"));
        // encode side: a frame that would serialise past the cap is refused
        let f = DaemonFrame::Error { session_id: None, message: big };
        assert!(encode_line(&f).unwrap_err().contains("64 MiB"));
        // a > 1 MiB (rpc cap) frame is fine on the daemon wire
        let f = DaemonFrame::Error { session_id: None, message: "y".repeat(RPC_MAX_FRAME_BYTES * 2) };
        let line = encode_line(&f).unwrap();
        assert!(decode_line::<DaemonFrame>(line.trim_end()).is_ok());
    }

    #[test]
    fn build_id_is_set_once_by_the_binary() {
        // The only caller in this test binary; `main` is the only one in `synaps`.
        set_build_id(" abc1234-dirty \n");
        assert_eq!(build_id(), "abc1234-dirty");
        set_build_id("later");
        assert_eq!(build_id(), "abc1234-dirty", "first call wins");
    }

    #[test]
    fn no_secret_bearing_fields_on_handshake_frames() {
        // Structural guard: the JSON of Welcome/Attached must not contain key-ish names.
        let w = serde_json::to_string(&Welcome {
            protocol_version: 1,
            daemon_version: "v".into(),
            pid: 1,
            profile: Some("p".into()),
            sessions: vec![meta()],
            progressive_tool_disclosure: false,
            generation: 1,
            daemon_build: String::new(),
        })
        .unwrap()
        .to_lowercase();
        for bad in ["token", "secret", "api_key", "apikey", "password", "credential"] {
            assert!(!w.contains(bad), "{bad} in Welcome");
        }
    }
}

#[cfg(test)]
mod float_roundtrip_tests {
    use super::*;

    /// L≡S (phase 4 G9): the session cost crosses the wire as an f64. Without
    /// serde_json's `float_roundtrip` the parse is one ulp short for values
    /// like `0.000105 + 0.000045` and the footer rounds `$0.0002` → `$0.0001`.
    #[test]
    fn conversation_digest_cost_survives_the_wire_exactly() {
        let cost = (10f64 / 1_000_000.0) * 3.0 + (5f64 / 1_000_000.0) * 15.0
            + ((10f64 / 1_000_000.0) * 3.0 + (1f64 / 1_000_000.0) * 15.0);
        let snap = ConversationSnapshot { cost, ..Default::default() };
        let d = ConversationDigest::of(&snap);
        let line = encode_line(&d).unwrap();
        let back: ConversationDigest = decode_line(line.trim_end()).unwrap();
        assert_eq!(back.cost.to_bits(), cost.to_bits(), "{line}");
        assert_eq!(format!("{:.4}", back.cost), format!("{:.4}", cost));
    }
}