agent_engine/session/
wire.rs

1//! Wire frames for the daemon UDS (PLAN-phase2 §2.9, S289 D-2 amended).
2//!
3//! Framing is `synaps rpc`'s: one JSON object per `\n`, at most
4//! [`DAEMON_MAX_FRAME_BYTES`] (64 MiB — distinct from rpc's 1 MiB because
5//! `Attached` carries the whole `api_messages`). Enforced symmetrically:
6//! `encode_line` refuses to build an oversize frame, `decode_line` and both
7//! readers refuse to accept one. `WireSessionEvent` is a serde mirror of
8//! `SessionEventWire` — `StreamEvent`/`TurnError`/`ExtensionLoaderEvent` are
9//! not `Serialize`, so they are mirrored variant-for-variant here and
10//! converted only at the socket boundary. In-process transports never touch
11//! this module.
12//!
13//! The one *lossy* mirror is `Conversation`: on the wire it is a
14//! [`ConversationDigest`] (len + hash + tokens/cost), never the messages.
15//! Full history travels only in `Attached` and `QueryResult{Messages}`;
16//! `SocketTransport` rebuilds the snapshot from its local mirror
17//! (`Attached` + `Stream(MessageHistory)`) and re-queries on a hash miss.
18//!
19//! Security: nothing in here carries a credential. `Welcome`/`Attached` are
20//! summaries; `ClientFrame`'s `Debug` redacts `Answer.value` and
21//! `Submit.text` so a frame can be traced by *type* without leaking bodies.
22
23use std::path::PathBuf;
24
25use serde::{Deserialize, Serialize};
26
27use super::types::*;
28use super::view::RuntimeView;
29use crate::{AgentEvent, LlmEvent, SessionEvent, StreamEvent, TurnError, TurnOutcome};
30
31/// Separate from `RPC_PROTOCOL_VERSION`; bump on any non-additive change.
32/// v2 (phase 3): `Set` became a struct variant, `Bye` gained a reason.
33/// v3 (E-P0): driver events (`DriverArmed`, `DriverRevoked`,
34/// `DriverTurnOutcome`); old clients skip unknown via `#[serde(other)]`.
35/// Protocol stays exact-match; only *binary* versions compare directionally
36/// (reload, §2.8).
37pub const PROTOCOL_VERSION: u32 = 3;
38/// Exact-match policy today (`min == max == PROTOCOL_VERSION`).
39pub const PROTOCOL_MIN: u32 = PROTOCOL_VERSION;
40pub const PROTOCOL_MAX: u32 = PROTOCOL_VERSION;
41/// rpc's cap, kept for reference; daemon frames use [`DAEMON_MAX_FRAME_BYTES`].
42pub const RPC_MAX_FRAME_BYTES: usize = agent_core::core::rpc_dispatch::MAX_FRAME_BYTES;
43/// Hard cap on one daemon frame, both directions (`Attached` ships the
44/// whole conversation; 64 MiB is well past any context window's JSON).
45pub const DAEMON_MAX_FRAME_BYTES: usize = 64 * 1024 * 1024;
46/// Alias used by the readers/writers of this protocol.
47pub const MAX_FRAME_BYTES: usize = DAEMON_MAX_FRAME_BYTES;
48/// Query ids at or above this are reserved for the transport/daemon and
49/// never surfaced to a client as `QueryResult`.
50pub const RESERVED_QUERY_ID_BASE: u64 = 1 << 63;
51/// `SocketTransport` re-fetches `api_messages` under this id on a digest miss.
52pub const DIGEST_RESYNC_QUERY_ID: u64 = RESERVED_QUERY_ID_BASE;
53/// The daemon's idle monitor probes `Status` under this id.
54pub const IDLE_PROBE_QUERY_ID: u64 = RESERVED_QUERY_ID_BASE + 1;
55/// `Checkpoint` replies `QueryResult{id: CHECKPOINT_QUERY_ID, {ok:true}}`
56/// so `reload.rs` can await it per session.
57pub const CHECKPOINT_QUERY_ID: u64 = RESERVED_QUERY_ID_BASE + 2;
58
59/// The binary version the daemon/client report in the handshake.
60pub fn binary_version() -> String {
61    env!("CARGO_PKG_VERSION").to_string()
62}
63
64// ── client → daemon ───────────────────────────────────────────────────────────
65
66/// Client → daemon. First frame on a connection MUST be `Hello`.
67#[derive(Serialize, Deserialize)]
68#[serde(tag = "type", rename_all = "snake_case")]
69pub enum ClientFrame {
70    Hello(Hello),
71    // ── lightweight control fast path: no session allocated, no Attach ──
72    Ping,
73    Sessions,
74    Shutdown {
75        #[serde(default)]
76        force: bool,
77    },
78    // ── session-scoped ──
79    Attach(Attach),
80    Cmd {
81        session_id: SessionId,
82        cmd: SessionCommand,
83    },
84    /// Control fast path (no session): checkpoint every session, write
85    /// reload-state, execv self. Reply: `Reloading` to every attached
86    /// client, `Bye` to this one. (C3)
87    Reload {
88        #[serde(default)]
89        now: bool,
90        #[serde(default)]
91        drain_secs: Option<u64>,
92        #[serde(default)]
93        exe: Option<PathBuf>,
94    },
95    /// Control fast path: `memstat::purge_arenas()` in the daemon (bench
96    /// hygiene). Reply: `Pong`. (C2)
97    Purge,
98    Bye,
99}
100
101impl std::fmt::Debug for ClientFrame {
102    /// Frame *types* only for anything that can carry user text or a secret
103    /// (`SessionCommand`'s own `Debug` redacts bodies without lengths).
104    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105        match self {
106            Self::Hello(h) => f.debug_tuple("Hello").field(h).finish(),
107            Self::Ping => f.write_str("Ping"),
108            Self::Sessions => f.write_str("Sessions"),
109            Self::Shutdown { force } => f.debug_struct("Shutdown").field("force", force).finish(),
110            Self::Attach(a) => f.debug_tuple("Attach").field(a).finish(),
111            Self::Cmd { session_id, cmd } => {
112                f.debug_struct("Cmd").field("session_id", session_id).field("cmd", cmd).finish()
113            }
114            Self::Reload { now, drain_secs, exe } => f
115                .debug_struct("Reload")
116                .field("now", now)
117                .field("drain_secs", drain_secs)
118                .field("exe", exe)
119                .finish(),
120            Self::Purge => f.write_str("Purge"),
121            Self::Bye => f.write_str("Bye"),
122        }
123    }
124}
125
126#[derive(Debug, Clone, Serialize, Deserialize)]
127pub struct Hello {
128    pub protocol_version: u32,
129    pub client: ClientMeta,
130    pub cwd: PathBuf,
131    pub client_version: String,
132    /// Client environment snapshot (secrets and `SYNAPS_CLIENT_*`/`SYNAPS_TUI_*`/
133    /// `SYNAPS_DAEMON_*`/`SYNAPS_MEM_TRACE*` stripped). `None` = in-process
134    /// (inherit process env). Legacy frames without this field decode as `None`.
135    #[serde(default)]
136    pub env: Option<super::types::SessionEnv>,
137    /// Names of env vars stripped as secrets at the client (T5). Names only,
138    /// never values. `serde(default)` → legacy frames decode as empty.
139    #[serde(default, skip_serializing_if = "Vec::is_empty")]
140    pub env_stripped: Vec<String>,
141    /// Set when re-connecting after `Reloading`/EOF (C3/A4).
142    #[serde(default)]
143    pub reconnect_of: Option<ClientReconnect>,
144}
145
146/// Who this connection used to be, so the daemon can restore ownership by
147/// reconnect order + `was_owner`.
148#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
149pub struct ClientReconnect {
150    pub previous_client: ClientId,
151    pub session_id: SessionId,
152    pub was_owner: bool,
153    pub generation: u64,
154}
155
156impl Hello {
157    pub fn new(kind: ClientKind) -> Self {
158        let (env, env_stripped) = super::types::capture_client_env();
159        Self {
160            protocol_version: PROTOCOL_VERSION,
161            client: ClientMeta {
162                terminal: std::env::var("TERM").ok(),
163                ..ClientMeta::new(kind)
164            },
165            cwd: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
166            client_version: binary_version(),
167            env: Some(env),
168            env_stripped,
169            reconnect_of: None,
170        }
171    }
172
173    /// Ask for `Full` (mirror) or `Digest` (digest + display tail) history.
174    pub fn with_history(mut self, history: HistoryMode) -> Self {
175        self.client.history = history;
176        self
177    }
178
179    /// Display items wanted in `Attached.display_tail` / `DisplayTail` queries.
180    pub fn with_tail_items(mut self, tail_items: usize) -> Self {
181        self.client.tail_items = tail_items;
182        self
183    }
184}
185
186#[derive(Debug, Clone, Serialize, Deserialize)]
187#[serde(tag = "attach", rename_all = "snake_case")]
188pub enum Attach {
189    Existing {
190        session_id: SessionId,
191        #[serde(default = "mirror")]
192        mode: AttachMode,
193    },
194    Create {
195        #[serde(default)]
196        config: SessionConfig,
197        #[serde(default = "mirror")]
198        mode: AttachMode,
199    },
200}
201
202fn mirror() -> AttachMode {
203    AttachMode::Mirror
204}
205
206// ── daemon → client ───────────────────────────────────────────────────────────
207
208#[derive(Debug, Serialize, Deserialize)]
209#[serde(tag = "type", rename_all = "snake_case")]
210#[allow(clippy::large_enum_variant)]
211pub enum DaemonFrame {
212    Welcome(Welcome),
213    /// Then the daemon closes.
214    Refused {
215        reason: RefuseReason,
216        message: String,
217    },
218    Pong {
219        pid: u32,
220        uptime_s: u64,
221        sessions: usize,
222    },
223    SessionList {
224        sessions: Vec<SessionMeta>,
225    },
226    Attached(AttachedWire),
227    Event(WireEnvelope),
228    Error {
229        session_id: Option<SessionId>,
230        message: String,
231    },
232    /// `Some(Reloading{..})` before exec; `None` otherwise (today).
233    Bye {
234        #[serde(default)]
235        reason: Option<ByeReason>,
236    },
237}
238
239#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
240#[serde(tag = "bye", rename_all = "snake_case")]
241pub enum ByeReason {
242    Reloading { generation: u64, retry_after_ms: u64 },
243    Shutdown,
244    Detached,
245}
246
247#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
248#[serde(tag = "reason", rename_all = "snake_case")]
249pub enum RefuseReason {
250    Version {
251        daemon_version: u32,
252        min: u32,
253        max: u32,
254    },
255    Protocol,
256    Busy,
257    UnknownSession,
258    Config,
259    /// `synaps daemon reload` version gate failed (C3).
260    ReloadRefused { why: String },
261}
262
263/// Auth-status *summary* only — never key material.
264#[derive(Debug, Clone, Serialize, Deserialize)]
265pub struct Welcome {
266    pub protocol_version: u32,
267    pub daemon_version: String,
268    pub pid: u32,
269    pub profile: Option<String>,
270    pub sessions: Vec<SessionMeta>,
271    pub progressive_tool_disclosure: bool,
272    /// Reload counter (starts at 1).
273    #[serde(default = "one")]
274    pub generation: u64,
275}
276
277fn one() -> u64 {
278    1
279}
280
281#[derive(Debug, Clone, Serialize, Deserialize)]
282pub struct WireEnvelope {
283    pub session_id: SessionId,
284    pub seq: u64,
285    pub ts: chrono::DateTime<chrono::Utc>,
286    pub event: WireSessionEvent,
287}
288
289impl From<Envelope> for WireEnvelope {
290    fn from(e: Envelope) -> Self {
291        Self {
292            session_id: e.session_id,
293            seq: e.seq,
294            ts: e.ts,
295            event: e.event.into(),
296        }
297    }
298}
299
300impl From<WireEnvelope> for Envelope {
301    fn from(e: WireEnvelope) -> Self {
302        Self {
303            session_id: e.session_id,
304            seq: e.seq,
305            ts: e.ts,
306            event: e.event.into(),
307        }
308    }
309}
310
311/// `AttachSnapshot` with a serialisable replay.
312#[derive(Debug, Clone, Serialize, Deserialize)]
313pub struct AttachedWire {
314    pub client: ClientId,
315    pub meta: SessionMeta,
316    pub view: RuntimeView,
317    pub conversation: ConversationSnapshot,
318    pub streaming: bool,
319    pub replay: Vec<WireEnvelope>,
320    pub pending_prompts: Vec<PromptRequest>,
321    pub clients: Vec<(ClientId, ClientKind)>,
322    #[serde(default)]
323    pub input_owner: Option<ClientId>,
324    #[serde(default)]
325    pub display_tail: Option<crate::session::display::DisplayTail>,
326}
327
328impl AttachedWire {
329    pub fn new(client: ClientId, s: AttachSnapshot) -> Self {
330        Self {
331            client,
332            meta: s.meta,
333            view: s.view,
334            conversation: s.conversation,
335            streaming: s.streaming,
336            replay: s.replay.into_iter().map(Into::into).collect(),
337            pending_prompts: s.pending_prompts,
338            clients: s.clients,
339            input_owner: s.input_owner,
340            display_tail: s.display_tail,
341        }
342    }
343
344    pub fn into_snapshot(self) -> (ClientId, AttachSnapshot) {
345        (
346            self.client,
347            AttachSnapshot {
348                meta: self.meta,
349                view: self.view,
350                conversation: self.conversation,
351                streaming: self.streaming,
352                replay: self.replay.into_iter().map(Into::into).collect(),
353                pending_prompts: self.pending_prompts,
354                clients: self.clients,
355                input_owner: self.input_owner,
356                display_tail: self.display_tail,
357            },
358        )
359    }
360}
361
362// ── conversation digest ───────────────────────────────────────────────────────
363
364/// `ConversationSnapshot` minus `api_messages`: what `Conversation` carries
365/// on the wire. `messages_hash` is FNV-1a over each message's JSON so a
366/// client can tell whether its local mirror is current.
367#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
368pub struct ConversationDigest {
369    #[serde(default)]
370    pub header: SessionHeader,
371    pub messages_len: usize,
372    pub messages_hash: u64,
373    pub tokens: ConversationTokens,
374    pub cost: f64,
375    pub abort_context: Option<String>,
376    pub queued_message: Option<String>,
377    pub pending_events_len: usize,
378    pub consecutive_auto_turns: u32,
379}
380
381/// FNV-1a 64 over the canonical JSON of every message (stable across
382/// binaries, unlike `DefaultHasher`).
383pub fn messages_hash(messages: &[crate::SharedMessage]) -> u64 {
384    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
385    let mut feed = |b: u8| {
386        h ^= b as u64;
387        h = h.wrapping_mul(0x0000_0100_0000_01b3);
388    };
389    for m in messages {
390        if let Ok(bytes) = serde_json::to_vec(&**m) {
391            for b in bytes {
392                feed(b);
393            }
394        }
395        feed(0x1f);
396    }
397    h
398}
399
400impl ConversationDigest {
401    pub fn of(s: &ConversationSnapshot) -> Self {
402        Self {
403            header: s.header.clone(),
404            messages_len: s.api_messages.len(),
405            messages_hash: s.messages_hash_memo.hash(&s.api_messages),
406            tokens: s.tokens.clone(),
407            cost: s.cost,
408            abort_context: s.abort_context.clone(),
409            queued_message: s.queued_message.clone(),
410            pending_events_len: s.pending_events_len,
411            consecutive_auto_turns: s.consecutive_auto_turns,
412        }
413    }
414
415    pub fn matches(&self, messages: &[crate::SharedMessage]) -> bool {
416        self.messages_len == messages.len() && self.messages_hash == messages_hash(messages)
417    }
418
419    /// Rebuild a snapshot around `messages` (the caller's mirror).
420    pub fn into_snapshot(self, api_messages: Vec<crate::SharedMessage>) -> ConversationSnapshot {
421        ConversationSnapshot {
422            header: self.header,
423            messages_len: self.messages_len,
424            api_messages,
425            tokens: self.tokens,
426            cost: self.cost,
427            abort_context: self.abort_context,
428            queued_message: self.queued_message,
429            pending_events_len: self.pending_events_len,
430            consecutive_auto_turns: self.consecutive_auto_turns,
431            messages_hash_memo: Default::default(),
432        }
433    }
434}
435
436// ── mirror of SessionEventWire (lossless except Conversation → digest) ────────
437
438#[derive(Debug, Clone, Serialize, Deserialize)]
439#[serde(tag = "ev", rename_all = "snake_case")]
440#[allow(clippy::large_enum_variant)] // mirrors SessionEventWire by design
441pub enum WireSessionEvent {
442    Stream { event: WireStreamEvent },
443    TurnStarted {
444        turn_baseline: usize,
445        trigger: TurnTrigger,
446        #[serde(default)]
447        user_text: Option<String>,
448    },
449    /// Digest only — see module docs.
450    Conversation { digest: ConversationDigest },
451    Prompt { request: PromptRequest },
452    PromptResolved { prompt_id: u64 },
453    External { event: crate::events::types::Event },
454    AutoTurnCapReached { cap: u32 },
455    Idle,
456    Steered { text: String, delivered: bool },
457    Dequeued { text: String },
458    SystemNotice { text: String },
459    LoaderProgress { event: WireLoaderEvent },
460    ExtensionNotification { extension_id: String, method: String, params: serde_json::Value },
461    SettingChanged { applied: SettingApplied },
462    QueryResult { id: u64, value: serde_json::Value },
463    Attached { client: ClientId, snapshot: AttachedWire },
464    ClientJoined { client: ClientId, kind: ClientKind },
465    ClientLeft { client: ClientId },
466    Ended { reason: EndReason },
467    Aborted { context_saved: bool },
468    Cleared { session_id: String },
469    CompactionStarted { source: String, disclosure: String },
470    CompactionApplied {
471        previous_session_id: String,
472        session_id: String,
473        chains_advanced: Vec<String>,
474        queued_restored: Option<String>,
475        msg_count: usize,
476    },
477    CompactionFailed { message: String, panicked: bool },
478    CompactionCancelled,
479    SubagentRows { rows: Vec<crate::runtime::subagent::SubagentDisplayRow> },
480    Resumed {
481        id: u64,
482        old_id: String,
483        new_id: String,
484        via: Option<String>,
485        clamp_notice: Option<String>,
486    },
487    InputOwnerChanged { from: Option<ClientId>, to: Option<ClientId>, reason: OwnerChangeReason },
488    Refused { client: ClientId, command: String, reason: String },
489    AttachRefused { message: String },
490    Lifecycle { lifecycle: SessionLifecycle },
491    Reloading { generation: u64, retry_after_ms: u64 },
492    /// (E-P0) Driver armed.
493    DriverArmed {
494        plugin_id: String,
495        run_id: String,
496        models: Vec<crate::extensions::session_driver::Selection>,
497        selection: crate::extensions::session_driver::Selection,
498        #[serde(default)]
499        deadline_ms: Option<u64>,
500        #[serde(default)]
501        notice: String,
502    },
503    /// (E-P0) Driver revoked.
504    DriverRevoked {
505        reason: String,
506        #[serde(default)]
507        undelivered_steering: Vec<String>,
508    },
509    /// (E-P0) Driver turn outcome.
510    DriverTurnOutcome {
511        outcome: crate::extensions::session_driver::Outcome,
512        selection: crate::extensions::session_driver::Selection,
513        #[serde(default)]
514        feedback: Option<String>,
515    },
516    /// (E-P7) Spend ceiling breached; turn cancelled and driver revoked.
517    CostCapReached {
518        scope: String,
519        cost: f64,
520        cap: f64,
521    },
522    /// Forward-compat: an additive variant from a newer daemon.
523    #[serde(other)]
524    Unknown,
525}
526
527#[derive(Debug, Clone, Serialize, Deserialize)]
528#[serde(tag = "kind", rename_all = "snake_case")]
529pub enum WireStreamEvent {
530    Llm(WireLlmEvent),
531    Session(WireSessionStreamEvent),
532    Agent(WireAgentEvent),
533}
534
535#[derive(Debug, Clone, Serialize, Deserialize)]
536#[serde(tag = "llm", rename_all = "snake_case")]
537pub enum WireLlmEvent {
538    Thinking { text: String },
539    Text { text: String },
540    ToolUseStart { tool_name: String, tool_id: String },
541    ToolUseDelta { tool_id: String, delta: String },
542    ToolUse { tool_name: String, tool_id: String, input: serde_json::Value },
543    ToolResult { tool_id: String, result: String },
544    ToolResultDelta { tool_id: String, delta: String },
545    ResponseStart,
546    ResponseReset,
547}
548
549#[derive(Debug, Clone, Serialize, Deserialize)]
550#[serde(tag = "session", rename_all = "snake_case")]
551pub enum WireSessionStreamEvent {
552    MessageHistory { messages: Vec<crate::SharedMessage> },
553    Usage {
554        input_tokens: u64,
555        output_tokens: u64,
556        cache_read_input_tokens: u64,
557        cache_creation_input_tokens: u64,
558        cache_creation_5m: Option<u64>,
559        cache_creation_1h: Option<u64>,
560        model: Option<String>,
561    },
562    Done,
563    Error { message: String, outcome: TurnOutcome },
564    Notice { text: String },
565}
566
567#[derive(Debug, Clone, Serialize, Deserialize)]
568#[serde(tag = "agent", rename_all = "snake_case")]
569pub enum WireAgentEvent {
570    SubagentStart { subagent_id: u64, agent_name: String, task_preview: String },
571    SubagentUpdate { subagent_id: u64, agent_name: String, status: String },
572    SubagentDone { subagent_id: u64, agent_name: String, result_preview: String, duration_secs: f64 },
573    SteeringDelivered { message: String },
574}
575
576#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
577pub struct WireLoadFailure {
578    pub plugin: String,
579    pub manifest_path: Option<PathBuf>,
580    pub reason: String,
581    pub hint: String,
582}
583
584#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
585#[serde(tag = "loader", rename_all = "snake_case")]
586pub enum WireLoaderEvent {
587    Started,
588    Loaded { plugin: String, loaded: usize, failed: usize },
589    Failed { failure: WireLoadFailure, loaded: usize, failed: usize },
590    Finished { loaded: Vec<String>, failed: Vec<WireLoadFailure> },
591}
592
593type LoadFailure = crate::extensions::manager::ExtensionLoadFailure;
594type LoaderEvent = crate::extensions::loader::ExtensionLoaderEvent;
595
596impl From<LoadFailure> for WireLoadFailure {
597    fn from(f: LoadFailure) -> Self {
598        Self { plugin: f.plugin, manifest_path: f.manifest_path, reason: f.reason, hint: f.hint }
599    }
600}
601impl From<WireLoadFailure> for LoadFailure {
602    fn from(f: WireLoadFailure) -> Self {
603        Self { plugin: f.plugin, manifest_path: f.manifest_path, reason: f.reason, hint: f.hint }
604    }
605}
606impl From<LoaderEvent> for WireLoaderEvent {
607    fn from(e: LoaderEvent) -> Self {
608        match e {
609            LoaderEvent::Started => Self::Started,
610            LoaderEvent::Loaded { plugin, loaded, failed } => Self::Loaded { plugin, loaded, failed },
611            LoaderEvent::Failed { failure, loaded, failed } => {
612                Self::Failed { failure: failure.into(), loaded, failed }
613            }
614            LoaderEvent::Finished { loaded, failed } => Self::Finished {
615                loaded,
616                failed: failed.into_iter().map(Into::into).collect(),
617            },
618        }
619    }
620}
621impl From<WireLoaderEvent> for LoaderEvent {
622    fn from(e: WireLoaderEvent) -> Self {
623        match e {
624            WireLoaderEvent::Started => Self::Started,
625            WireLoaderEvent::Loaded { plugin, loaded, failed } => Self::Loaded { plugin, loaded, failed },
626            WireLoaderEvent::Failed { failure, loaded, failed } => {
627                Self::Failed { failure: failure.into(), loaded, failed }
628            }
629            WireLoaderEvent::Finished { loaded, failed } => Self::Finished {
630                loaded,
631                failed: failed.into_iter().map(Into::into).collect(),
632            },
633        }
634    }
635}
636
637impl From<StreamEvent> for WireStreamEvent {
638    fn from(e: StreamEvent) -> Self {
639        match e {
640            StreamEvent::Llm(l) => Self::Llm(match l {
641                LlmEvent::Thinking(text) => WireLlmEvent::Thinking { text },
642                LlmEvent::Text(text) => WireLlmEvent::Text { text },
643                LlmEvent::ToolUseStart { tool_name, tool_id } => WireLlmEvent::ToolUseStart { tool_name, tool_id },
644                LlmEvent::ToolUseDelta { tool_id, delta } => WireLlmEvent::ToolUseDelta { tool_id, delta },
645                LlmEvent::ToolUse { tool_name, tool_id, input } => WireLlmEvent::ToolUse { tool_name, tool_id, input },
646                LlmEvent::ToolResult { tool_id, result } => WireLlmEvent::ToolResult { tool_id, result },
647                LlmEvent::ToolResultDelta { tool_id, delta } => WireLlmEvent::ToolResultDelta { tool_id, delta },
648                LlmEvent::ResponseStart => WireLlmEvent::ResponseStart,
649                LlmEvent::ResponseReset => WireLlmEvent::ResponseReset,
650            }),
651            StreamEvent::Session(s) => Self::Session(match s {
652                SessionEvent::MessageHistory(messages) => WireSessionStreamEvent::MessageHistory { messages },
653                SessionEvent::Usage {
654                    input_tokens,
655                    output_tokens,
656                    cache_read_input_tokens,
657                    cache_creation_input_tokens,
658                    cache_creation_5m,
659                    cache_creation_1h,
660                    model,
661                } => WireSessionStreamEvent::Usage {
662                    input_tokens,
663                    output_tokens,
664                    cache_read_input_tokens,
665                    cache_creation_input_tokens,
666                    cache_creation_5m,
667                    cache_creation_1h,
668                    model,
669                },
670                SessionEvent::Done => WireSessionStreamEvent::Done,
671                SessionEvent::Error(TurnError { message, outcome }) => WireSessionStreamEvent::Error { message, outcome },
672                SessionEvent::Notice(text) => WireSessionStreamEvent::Notice { text },
673                // merge(112): ContextHeadCheckpoint is actor-internal (Wall 1); never crosses the wire.
674                // Drop it silently — the actor handles persistence before forwarding.
675                SessionEvent::ContextHeadCheckpoint { .. } => WireSessionStreamEvent::Done,
676            }),
677            StreamEvent::Agent(a) => Self::Agent(match a {
678                AgentEvent::SubagentStart { subagent_id, agent_name, task_preview } => {
679                    WireAgentEvent::SubagentStart { subagent_id, agent_name, task_preview }
680                }
681                AgentEvent::SubagentUpdate { subagent_id, agent_name, status } => {
682                    WireAgentEvent::SubagentUpdate { subagent_id, agent_name, status }
683                }
684                AgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs } => {
685                    WireAgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs }
686                }
687                AgentEvent::SteeringDelivered { message } => WireAgentEvent::SteeringDelivered { message },
688            }),
689        }
690    }
691}
692
693impl From<WireStreamEvent> for StreamEvent {
694    fn from(e: WireStreamEvent) -> Self {
695        match e {
696            WireStreamEvent::Llm(l) => Self::Llm(match l {
697                WireLlmEvent::Thinking { text } => LlmEvent::Thinking(text),
698                WireLlmEvent::Text { text } => LlmEvent::Text(text),
699                WireLlmEvent::ToolUseStart { tool_name, tool_id } => LlmEvent::ToolUseStart { tool_name, tool_id },
700                WireLlmEvent::ToolUseDelta { tool_id, delta } => LlmEvent::ToolUseDelta { tool_id, delta },
701                WireLlmEvent::ToolUse { tool_name, tool_id, input } => LlmEvent::ToolUse { tool_name, tool_id, input },
702                WireLlmEvent::ToolResult { tool_id, result } => LlmEvent::ToolResult { tool_id, result },
703                WireLlmEvent::ToolResultDelta { tool_id, delta } => LlmEvent::ToolResultDelta { tool_id, delta },
704                WireLlmEvent::ResponseStart => LlmEvent::ResponseStart,
705                WireLlmEvent::ResponseReset => LlmEvent::ResponseReset,
706            }),
707            WireStreamEvent::Session(s) => Self::Session(match s {
708                WireSessionStreamEvent::MessageHistory { messages } => SessionEvent::MessageHistory(messages),
709                WireSessionStreamEvent::Usage {
710                    input_tokens,
711                    output_tokens,
712                    cache_read_input_tokens,
713                    cache_creation_input_tokens,
714                    cache_creation_5m,
715                    cache_creation_1h,
716                    model,
717                } => SessionEvent::Usage {
718                    input_tokens,
719                    output_tokens,
720                    cache_read_input_tokens,
721                    cache_creation_input_tokens,
722                    cache_creation_5m,
723                    cache_creation_1h,
724                    model,
725                },
726                WireSessionStreamEvent::Done => SessionEvent::Done,
727                WireSessionStreamEvent::Error { message, outcome } => SessionEvent::Error(TurnError { message, outcome }),
728                WireSessionStreamEvent::Notice { text } => SessionEvent::Notice(text),
729            }),
730            WireStreamEvent::Agent(a) => Self::Agent(match a {
731                WireAgentEvent::SubagentStart { subagent_id, agent_name, task_preview } => {
732                    AgentEvent::SubagentStart { subagent_id, agent_name, task_preview }
733                }
734                WireAgentEvent::SubagentUpdate { subagent_id, agent_name, status } => {
735                    AgentEvent::SubagentUpdate { subagent_id, agent_name, status }
736                }
737                WireAgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs } => {
738                    AgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs }
739                }
740                WireAgentEvent::SteeringDelivered { message } => AgentEvent::SteeringDelivered { message },
741            }),
742        }
743    }
744}
745
746impl From<SessionEventWire> for WireSessionEvent {
747    fn from(e: SessionEventWire) -> Self {
748        use SessionEventWire as S;
749        match e {
750            S::Stream(ev) => Self::Stream { event: ev.into() },
751            S::TurnStarted { turn_baseline, trigger, user_text } => {
752                Self::TurnStarted { turn_baseline, trigger, user_text }
753            }
754            S::Conversation(snapshot) => Self::Conversation { digest: ConversationDigest::of(&snapshot) },
755            S::Prompt(request) => Self::Prompt { request },
756            S::PromptResolved { prompt_id } => Self::PromptResolved { prompt_id },
757            S::External(event) => Self::External { event },
758            S::AutoTurnCapReached { cap } => Self::AutoTurnCapReached { cap },
759            S::Idle => Self::Idle,
760            S::Steered { text, delivered } => Self::Steered { text, delivered },
761            S::Dequeued { text } => Self::Dequeued { text },
762            S::SystemNotice(text) => Self::SystemNotice { text },
763            S::LoaderProgress(ev) => Self::LoaderProgress { event: ev.into() },
764            S::ExtensionNotification { extension_id, method, params } => {
765                Self::ExtensionNotification { extension_id, method, params }
766            }
767            S::SettingChanged(applied) => Self::SettingChanged { applied },
768            S::QueryResult { id, value } => Self::QueryResult { id, value },
769            S::Attached { client, snapshot } => Self::Attached { client, snapshot: AttachedWire::new(client, snapshot) },
770            S::ClientJoined { client, kind } => Self::ClientJoined { client, kind },
771            S::ClientLeft { client } => Self::ClientLeft { client },
772            S::Ended { reason } => Self::Ended { reason },
773            S::Aborted { context_saved } => Self::Aborted { context_saved },
774            S::Cleared { session_id } => Self::Cleared { session_id },
775            S::CompactionStarted { source, disclosure } => Self::CompactionStarted { source, disclosure },
776            S::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count } => {
777                Self::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count }
778            }
779            S::CompactionFailed { message, panicked } => Self::CompactionFailed { message, panicked },
780            S::CompactionCancelled => Self::CompactionCancelled,
781            S::SubagentRows(rows) => Self::SubagentRows { rows },
782            S::Resumed { id, old_id, new_id, via, clamp_notice } => {
783                Self::Resumed { id, old_id, new_id, via, clamp_notice }
784            }
785            S::InputOwnerChanged { from, to, reason } => Self::InputOwnerChanged { from, to, reason },
786            S::Refused { client, command, reason } => Self::Refused { client, command, reason },
787            S::AttachRefused { message } => Self::AttachRefused { message },
788            S::Lifecycle(lifecycle) => Self::Lifecycle { lifecycle },
789            S::Reloading { generation, retry_after_ms } => Self::Reloading { generation, retry_after_ms },
790            S::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice } => {
791                Self::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice }
792            }
793            S::DriverRevoked { reason, undelivered_steering } => {
794                Self::DriverRevoked { reason, undelivered_steering }
795            }
796            S::DriverTurnOutcome { outcome, selection, feedback } => {
797                Self::DriverTurnOutcome { outcome, selection, feedback }
798            }
799            S::CostCapReached { scope, cost, cap } => {
800                Self::CostCapReached { scope, cost, cap }
801            }
802        }
803    }
804}
805
806impl From<WireSessionEvent> for SessionEventWire {
807    fn from(e: WireSessionEvent) -> Self {
808        use WireSessionEvent as W;
809        match e {
810            W::Stream { event } => Self::Stream(event.into()),
811            W::TurnStarted { turn_baseline, trigger, user_text } => {
812                Self::TurnStarted { turn_baseline, trigger, user_text }
813            }
814            // Messages are not on the wire; `SocketTransport` fills them from
815            // its mirror before handing the envelope up.
816            W::Conversation { digest } => Self::Conversation(digest.into_snapshot(Vec::new())),
817            W::Prompt { request } => Self::Prompt(request),
818            W::PromptResolved { prompt_id } => Self::PromptResolved { prompt_id },
819            W::External { event } => Self::External(event),
820            W::AutoTurnCapReached { cap } => Self::AutoTurnCapReached { cap },
821            W::Idle => Self::Idle,
822            W::Steered { text, delivered } => Self::Steered { text, delivered },
823            W::Dequeued { text } => Self::Dequeued { text },
824            W::SystemNotice { text } => Self::SystemNotice(text),
825            W::LoaderProgress { event } => Self::LoaderProgress(event.into()),
826            W::ExtensionNotification { extension_id, method, params } => {
827                Self::ExtensionNotification { extension_id, method, params }
828            }
829            W::SettingChanged { applied } => Self::SettingChanged(applied),
830            W::QueryResult { id, value } => Self::QueryResult { id, value },
831            W::Attached { snapshot, .. } => {
832                let (client, snapshot) = snapshot.into_snapshot();
833                Self::Attached { client, snapshot }
834            }
835            W::ClientJoined { client, kind } => Self::ClientJoined { client, kind },
836            W::ClientLeft { client } => Self::ClientLeft { client },
837            W::Ended { reason } => Self::Ended { reason },
838            W::Aborted { context_saved } => Self::Aborted { context_saved },
839            W::Cleared { session_id } => Self::Cleared { session_id },
840            W::CompactionStarted { source, disclosure } => Self::CompactionStarted { source, disclosure },
841            W::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count } => {
842                Self::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count }
843            }
844            W::CompactionFailed { message, panicked } => Self::CompactionFailed { message, panicked },
845            W::CompactionCancelled => Self::CompactionCancelled,
846            W::SubagentRows { rows } => Self::SubagentRows(rows),
847            W::Resumed { id, old_id, new_id, via, clamp_notice } => {
848                Self::Resumed { id, old_id, new_id, via, clamp_notice }
849            }
850            W::InputOwnerChanged { from, to, reason } => Self::InputOwnerChanged { from, to, reason },
851            W::Refused { client, command, reason } => Self::Refused { client, command, reason },
852            W::AttachRefused { message } => Self::AttachRefused { message },
853            W::Lifecycle { lifecycle } => Self::Lifecycle(lifecycle),
854            W::Reloading { generation, retry_after_ms } => Self::Reloading { generation, retry_after_ms },
855            W::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice } => {
856                Self::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice }
857            }
858            W::DriverRevoked { reason, undelivered_steering } => {
859                Self::DriverRevoked { reason, undelivered_steering }
860            }
861            W::DriverTurnOutcome { outcome, selection, feedback } => {
862                Self::DriverTurnOutcome { outcome, selection, feedback }
863            }
864            W::CostCapReached { scope, cost, cap } => {
865                Self::CostCapReached { scope, cost, cap }
866            }
867            W::Unknown => Self::SystemNotice("unknown event from a newer daemon (ignored)".into()),
868        }
869    }
870}
871
872// ── framing helpers ───────────────────────────────────────────────────────────
873
874/// Human-readable form of the cap for error frames.
875pub fn frame_limit_msg() -> String {
876    format!("frame exceeds {} MiB limit", DAEMON_MAX_FRAME_BYTES / (1024 * 1024))
877}
878
879/// Encode one frame as a single line (no embedded newline: serde_json
880/// escapes them). Refuses to produce a line over [`DAEMON_MAX_FRAME_BYTES`].
881pub fn encode_line<T: Serialize>(frame: &T) -> Result<String, String> {
882    let mut s = serde_json::to_string(frame).map_err(|e| e.to_string())?;
883    if s.len() + 1 > DAEMON_MAX_FRAME_BYTES {
884        return Err(frame_limit_msg());
885    }
886    s.push('\n');
887    Ok(s)
888}
889
890/// Decode one line, enforcing [`DAEMON_MAX_FRAME_BYTES`].
891pub fn decode_line<'a, T: Deserialize<'a>>(line: &'a str) -> Result<T, String> {
892    if line.len() > DAEMON_MAX_FRAME_BYTES {
893        return Err(frame_limit_msg());
894    }
895    serde_json::from_str(line).map_err(|e| e.to_string())
896}
897
898#[cfg(test)]
899mod tests {
900    use super::*;
901    use crate::BudgetDimension;
902    use std::sync::Arc;
903
904    fn view() -> RuntimeView {
905        crate::session::handle::echo::view_for()
906    }
907
908    fn meta() -> SessionMeta {
909        crate::session::handle::echo::meta_for(&SessionId::from("wire-1"))
910    }
911
912    fn prompt() -> PromptRequest {
913        PromptRequest {
914            id: 3,
915            kind: PromptKind::Secret,
916            title: "sudo".into(),
917            prompt: "password".into(),
918            raised_at: chrono::Utc::now(),
919        }
920    }
921
922    fn conv() -> ConversationSnapshot {
923        ConversationSnapshot {
924            header: SessionHeader {
925                id: "wire-1".into(),
926                title: "t".into(),
927                name: None,
928                model: "m".into(),
929                thinking_level: "off".into(),
930                system_prompt: Some("sys".into()),
931                created_at: chrono::Utc::now(),
932                updated_at: chrono::Utc::now(),
933                parent_session: Some("p".into()),
934            },
935            api_messages: vec![Arc::new(serde_json::json!({"role":"user","content":"hi"}))],
936            messages_len: 1,
937            tokens: ConversationTokens { input: 1, output: 2, cache_read: 3, cache_creation: 4 },
938            cost: 0.5,
939            abort_context: Some("ctx".into()),
940            queued_message: None,
941            pending_events_len: 2,
942            consecutive_auto_turns: 1,
943            messages_hash_memo: Default::default(),
944        }
945    }
946
947    /// One snapshot fanned out to N socket clients is hashed ONCE (the memo
948    /// is shared by its clones), and a snapshot whose messages were replaced
949    /// afterwards is re-hashed, never served a stale hash.
950    #[test]
951    fn conversation_digest_hash_is_computed_once_and_never_stale() {
952        let a = conv();
953        let b = a.clone();
954        assert_eq!(ConversationDigest::of(&a), ConversationDigest::of(&b));
955        assert!(a.messages_hash_memo.0.lock().unwrap().is_some(), "memo shared");
956        assert_eq!(
957            ConversationDigest::of(&b).messages_hash,
958            messages_hash(&a.api_messages)
959        );
960        let mut c = a.clone();
961        c.api_messages = Vec::new();
962        assert_eq!(ConversationDigest::of(&c).messages_hash, messages_hash(&[]));
963        let mut d = a.clone();
964        d.api_messages.push(Arc::new(serde_json::json!({"role":"assistant","content":"yo"})));
965        assert_eq!(
966            ConversationDigest::of(&d).messages_hash,
967            messages_hash(&d.api_messages)
968        );
969    }
970
971    fn env(event: SessionEventWire) -> Envelope {
972        Envelope { session_id: SessionId::from("wire-1"), seq: 9, ts: chrono::Utc::now(), event }
973    }
974
975    fn stream_fixtures() -> Vec<StreamEvent> {
976        vec![
977            StreamEvent::Llm(LlmEvent::Thinking("t".into())),
978            StreamEvent::Llm(LlmEvent::Text("x\ny".into())),
979            StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name: "bash".into(), tool_id: "id1".into() }),
980            StreamEvent::Llm(LlmEvent::ToolUseDelta { tool_id: "id1".into(), delta: "{\"c".into() }),
981            StreamEvent::Llm(LlmEvent::ToolUse {
982                tool_name: "bash".into(),
983                tool_id: "id1".into(),
984                input: serde_json::json!({"command":"pwd"}),
985            }),
986            StreamEvent::Llm(LlmEvent::ToolResult { tool_id: "id1".into(), result: "/tmp".into() }),
987            StreamEvent::Llm(LlmEvent::ToolResultDelta { tool_id: "id1".into(), delta: "/t".into() }),
988            StreamEvent::Session(SessionEvent::MessageHistory(vec![Arc::new(serde_json::json!({"role":"user","content":"a"}))])),
989            StreamEvent::Session(SessionEvent::Usage {
990                input_tokens: 1,
991                output_tokens: 2,
992                cache_read_input_tokens: 3,
993                cache_creation_input_tokens: 4,
994                cache_creation_5m: Some(5),
995                cache_creation_1h: None,
996                model: Some("m".into()),
997            }),
998            StreamEvent::Session(SessionEvent::Done),
999            StreamEvent::Session(SessionEvent::Error(TurnError::provider("boom", "auth_error", "turn-1-0"))),
1000            StreamEvent::Session(SessionEvent::Error(TurnError::budget(BudgetDimension::ToolCalls))),
1001            StreamEvent::Session(SessionEvent::Error(TurnError { message: "c".into(), outcome: TurnOutcome::Canceled })),
1002            StreamEvent::Session(SessionEvent::Notice("n".into())),
1003            StreamEvent::Agent(AgentEvent::SubagentStart { subagent_id: 1, agent_name: "a".into(), task_preview: "p".into() }),
1004            StreamEvent::Agent(AgentEvent::SubagentUpdate { subagent_id: 1, agent_name: "a".into(), status: "s".into() }),
1005            StreamEvent::Agent(AgentEvent::SubagentDone {
1006                subagent_id: 1,
1007                agent_name: "a".into(),
1008                result_preview: "r".into(),
1009                duration_secs: 1.5,
1010            }),
1011            StreamEvent::Agent(AgentEvent::SteeringDelivered { message: "m".into() }),
1012        ]
1013    }
1014
1015    fn fixtures() -> Vec<SessionEventWire> {
1016        use SessionEventWire as S;
1017        let loader = crate::extensions::loader::ExtensionLoaderEvent::Finished {
1018            loaded: vec!["p".into()],
1019            failed: vec![LoadFailure {
1020                plugin: "bad".into(),
1021                manifest_path: Some(PathBuf::from("/x/manifest.json")),
1022                reason: "oops".into(),
1023                hint: "fix".into(),
1024            }],
1025        };
1026        let mut v: Vec<S> = stream_fixtures().into_iter().map(S::Stream).collect();
1027        v.extend([
1028            S::TurnStarted { turn_baseline: 4, trigger: TurnTrigger::EventAuto, user_text: None },
1029            S::TurnStarted { turn_baseline: 5, trigger: TurnTrigger::QueuedAuto, user_text: Some("q".into()) },
1030            S::TurnStarted { turn_baseline: 6, trigger: TurnTrigger::DriverAuto, user_text: None },
1031            S::Conversation(conv()),
1032            S::Prompt(prompt()),
1033            S::PromptResolved { prompt_id: 3 },
1034            S::External(crate::events::types::Event::simple("cli", "hello", None)),
1035            S::AutoTurnCapReached { cap: 5 },
1036            S::Idle,
1037            S::Steered { text: "s".into(), delivered: true },
1038            S::Dequeued { text: "d".into() },
1039            S::SystemNotice("sys".into()),
1040            S::LoaderProgress(loader),
1041            S::ExtensionNotification { extension_id: "e".into(), method: "widget.upsert".into(), params: serde_json::json!({"a":1}) },
1042            S::SettingChanged(SettingApplied {
1043                id: 3,
1044                setting: "model".into(),
1045                ok: true,
1046                message: None,
1047                view: view(),
1048                clamp: Some(ReasoningClampWire { from: "ultra".into(), to: "high".into() }),
1049            }),
1050            S::QueryResult { id: 1, value: serde_json::json!([1, 2]) },
1051            S::Attached {
1052                client: ClientId(2),
1053                snapshot: AttachSnapshot {
1054                    meta: meta(),
1055                    view: view(),
1056                    conversation: conv(),
1057                    streaming: true,
1058                    replay: vec![env(S::Stream(StreamEvent::Llm(LlmEvent::Text("partial".into()))))],
1059                    pending_prompts: vec![prompt()],
1060                    clients: vec![(ClientId(1), ClientKind::Tui)],
1061                    input_owner: Some(ClientId(1)),
1062                    display_tail: None,
1063                },
1064            },
1065            S::ClientJoined { client: ClientId(1), kind: ClientKind::Attach },
1066            S::ClientLeft { client: ClientId(1) },
1067            S::Ended { reason: EndReason::HostShutdown },
1068            S::Aborted { context_saved: true },
1069            S::Cleared { session_id: "new-1".into() },
1070            S::CompactionStarted { source: "manual".into(), disclosure: "disc".into() },
1071            S::CompactionApplied {
1072                previous_session_id: "old".into(),
1073                session_id: "new".into(),
1074                chains_advanced: vec!["c1".into()],
1075                queued_restored: Some("q".into()),
1076                msg_count: 12,
1077            },
1078            S::CompactionFailed { message: "boom".into(), panicked: false },
1079            S::CompactionCancelled,
1080            S::SubagentRows(vec![crate::runtime::subagent::SubagentDisplayRow {
1081                subagent_id: 1,
1082                agent_name: "a".into(),
1083                status: crate::runtime::subagent::SubagentStatus::Failed("x".into()),
1084                cancel_requested: false,
1085                elapsed_secs: 1.25,
1086                finished_elapsed: Some(std::time::Duration::from_millis(1250)),
1087                step: "$ cargo test".into(),
1088                tools: 3,
1089            }]),
1090            S::Resumed { id: 2, old_id: "o".into(), new_id: "n".into(), via: Some("name".into()), clamp_notice: None },
1091            S::InputOwnerChanged { from: Some(ClientId(1)), to: Some(ClientId(2)), reason: OwnerChangeReason::Takeover },
1092            S::Refused { client: ClientId(2), command: "submit".into(), reason: "input owned by client #1".into() },
1093            S::AttachRefused { message: "parked".into() },
1094            S::Lifecycle(SessionLifecycle::Parked),
1095            S::Reloading { generation: 2, retry_after_ms: 500 },
1096            S::DriverArmed {
1097                plugin_id: "autonomous".into(),
1098                run_id: "run-1".into(),
1099                models: vec![
1100                    crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1101                    crate::extensions::session_driver::Selection { model: "m2".into(), effort: "low".into() },
1102                ],
1103                selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1104                deadline_ms: Some(60_000),
1105                notice: "armed".into(),
1106            },
1107            S::DriverRevoked { reason: "canceled".into(), undelivered_steering: vec!["steer1".into()] },
1108            S::DriverTurnOutcome {
1109                outcome: crate::extensions::session_driver::Outcome::Success,
1110                selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1111                feedback: Some("changed".into()),
1112            },
1113        ]);
1114        v
1115    }
1116
1117    #[test]
1118    fn wire_roundtrip_every_variant() {
1119        let all = fixtures();
1120        // 36 non-Stream variants (TurnStarted thrice = 3; +3 driver events) + 18 StreamEvent leaves.
1121        assert_eq!(all.len(), 36 + 18);
1122        for ev in all {
1123            let is_conv = matches!(ev, SessionEventWire::Conversation(_));
1124            let e = env(ev);
1125            let before = format!("{e:?}");
1126            let wire: WireEnvelope = e.into();
1127            let line = encode_line(&wire).unwrap();
1128            assert_eq!(line.matches('\n').count(), 1, "one line per frame");
1129            let back: WireEnvelope = decode_line(line.trim_end()).unwrap();
1130            let after = format!("{:?}", Envelope::from(back));
1131            if is_conv {
1132                // Digest on the wire: everything but the messages survives.
1133                assert!(!line.contains("api_messages"), "{line}");
1134                assert!(after.contains("api_messages: []"), "{after}");
1135                assert_eq!(before.replace("api_messages: [Object {\"content\": String(\"hi\"), \"role\": String(\"user\")}]", "api_messages: []"), after);
1136            } else {
1137                assert_eq!(before, after);
1138            }
1139        }
1140    }
1141
1142    #[test]
1143    fn hello_history_fields_default_and_roundtrip() {
1144        // Old peers omit the fields: Full / 120.
1145        let legacy = r#"{"kind":"tui","terminal":null,"instance":"i"}"#;
1146        let m: ClientMeta = serde_json::from_str(legacy).unwrap();
1147        assert_eq!(m.history, HistoryMode::Full);
1148        assert_eq!(m.tail_items, DEFAULT_TAIL_ITEMS);
1149        let h = Hello::new(ClientKind::Tui).with_history(HistoryMode::Digest).with_tail_items(9);
1150        let line = encode_line(&ClientFrame::Hello(h)).unwrap();
1151        assert!(line.contains("\"history\":\"digest\""), "{line}");
1152        assert!(line.contains("\"tail_items\":9"), "{line}");
1153        match decode_line::<ClientFrame>(line.trim_end()).unwrap() {
1154            ClientFrame::Hello(h) => {
1155                assert_eq!(h.client.history, HistoryMode::Digest);
1156                assert_eq!(h.client.tail_items, 9);
1157            }
1158            other => panic!("{other:?}"),
1159        }
1160        // Digest snapshots keep messages_len even with api_messages empty.
1161        let d = ConversationDigest::of(&conv());
1162        let snap = d.into_snapshot(Vec::new());
1163        assert_eq!(snap.messages_len, 1);
1164        assert!(snap.api_messages.is_empty());
1165        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();
1166        assert_eq!(legacy.messages_len, 0);
1167    }
1168
1169    #[test]
1170    fn conversation_digest_roundtrips_and_detects_drift() {
1171        let c = conv();
1172        let d = ConversationDigest::of(&c);
1173        assert_eq!(d.messages_len, 1);
1174        assert!(d.matches(&c.api_messages));
1175        assert!(!d.matches(&[]));
1176        let mut other = c.api_messages.clone();
1177        other.push(Arc::new(serde_json::json!({"role":"assistant","content":"x"})));
1178        assert!(!d.matches(&other));
1179        let back = d.clone().into_snapshot(c.api_messages.clone());
1180        assert_eq!(ConversationDigest::of(&back), d);
1181        // stable across runs/binaries: FNV-1a offset basis for the empty list
1182        assert_eq!(messages_hash(&[]), 0xcbf2_9ce4_8422_2325);
1183        assert_ne!(messages_hash(&c.api_messages), messages_hash(&[]));
1184    }
1185
1186    #[test]
1187    fn unknown_event_variant_tolerated() {
1188        let v: WireSessionEvent = serde_json::from_str(r#"{"ev":"from_the_future","x":1}"#).unwrap();
1189        assert!(matches!(v, WireSessionEvent::Unknown));
1190        assert!(matches!(SessionEventWire::from(v), SessionEventWire::SystemNotice(_)));
1191    }
1192
1193    #[test]
1194    fn driver_armed_round_trips_and_old_client_skips() {
1195        // A v3 daemon emits DriverArmed; a v2 client decodes it as Unknown.
1196        let armed = WireSessionEvent::DriverArmed {
1197            plugin_id: "autonomous".into(),
1198            run_id: "run-42".into(),
1199            models: vec![
1200                crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1201            ],
1202            selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1203            deadline_ms: Some(120_000),
1204            notice: "armed and ready".into(),
1205        };
1206        let json = serde_json::to_string(&armed).unwrap();
1207        // v3 client: exact round-trip.
1208        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
1209        assert!(matches!(&back, WireSessionEvent::DriverArmed { plugin_id, .. } if plugin_id == "autonomous"));
1210        // Simulated v2 client: rename tag so it hits Unknown.
1211        let old_json = json.replace(r#""ev":"driver_armed""#, r#""ev":"v2_unknown_event""#);
1212        let old: WireSessionEvent = serde_json::from_str(&old_json).unwrap();
1213        assert!(matches!(old, WireSessionEvent::Unknown));
1214    }
1215
1216    #[test]
1217    fn driver_revoked_and_outcome_round_trip() {
1218        let revoked = WireSessionEvent::DriverRevoked {
1219            reason: "canceled".into(),
1220            undelivered_steering: vec!["steer1".into(), "steer2".into()],
1221        };
1222        let json = serde_json::to_string(&revoked).unwrap();
1223        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
1224        assert!(matches!(&back, WireSessionEvent::DriverRevoked { reason, undelivered_steering }
1225            if reason == "canceled" && undelivered_steering.len() == 2));
1226
1227        let outcome = WireSessionEvent::DriverTurnOutcome {
1228            outcome: crate::extensions::session_driver::Outcome::Success,
1229            selection: crate::extensions::session_driver::Selection { model: "m".into(), effort: "high".into() },
1230            feedback: Some("changed".into()),
1231        };
1232        let json = serde_json::to_string(&outcome).unwrap();
1233        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
1234        assert!(matches!(&back, WireSessionEvent::DriverTurnOutcome { feedback: Some(f), .. } if f == "changed"));
1235    }
1236
1237    #[test]
1238    fn driver_revoked_defaults_tolerate_missing_fields() {
1239        // Old daemon or minimal frame: undelivered_steering missing → empty vec.
1240        let minimal = r#"{"ev":"driver_revoked","reason":"timeout"}"#;
1241        let v: WireSessionEvent = serde_json::from_str(minimal).unwrap();
1242        assert!(matches!(&v, WireSessionEvent::DriverRevoked { reason, undelivered_steering }
1243            if reason == "timeout" && undelivered_steering.is_empty()));
1244    }
1245
1246    #[test]
1247    fn client_frames_roundtrip_and_daemon_frames_roundtrip() {
1248        let frames = vec![
1249            ClientFrame::Hello(Hello::new(ClientKind::Attach)),
1250            ClientFrame::Hello(Hello::new(ClientKind::Tui).with_history(HistoryMode::Digest).with_tail_items(7)),
1251            ClientFrame::Ping,
1252            ClientFrame::Sessions,
1253            ClientFrame::Shutdown { force: true },
1254            ClientFrame::Attach(Attach::Existing { session_id: "s".into(), mode: AttachMode::Mirror }),
1255            ClientFrame::Attach(Attach::Create { config: SessionConfig::default(), mode: AttachMode::Takeover }),
1256            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Submit { text: "hi".into(), attachments: vec![] } },
1257            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Answer { prompt_id: 1, value: Some("pw".into()) } },
1258            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::EngineCommand { id: 4, name: "model".into(), arg: "x".into() } },
1259            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Query { id: 5, query: SessionQuery::Status } },
1260            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Set { id: 6, setting: SessionSetting::ReloadPrompt } },
1261            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Checkpoint { reason: CheckpointReason::Reload } },
1262            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::KeepWarm { on: true } },
1263            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::DriverStart {
1264                plugin: "autonomous".into(), command: "auto".into(), arg: "start -- do stuff".into(),
1265            }},
1266            ClientFrame::Reload { now: true, drain_secs: Some(3), exe: Some(PathBuf::from("/bin/synaps")) },
1267            ClientFrame::Purge,
1268            ClientFrame::Bye,
1269        ];
1270        for f in frames {
1271            let line = encode_line(&f).unwrap();
1272            let back: ClientFrame = decode_line(line.trim_end()).unwrap();
1273            assert_eq!(serde_json::to_string(&back).unwrap(), line.trim_end());
1274        }
1275        let frames = vec![
1276            DaemonFrame::Welcome(Welcome {
1277                protocol_version: PROTOCOL_VERSION,
1278                daemon_version: "0.9.0".into(),
1279                pid: 1,
1280                profile: None,
1281                sessions: vec![meta()],
1282                progressive_tool_disclosure: true,
1283                generation: 1,
1284            }),
1285            DaemonFrame::Refused { reason: RefuseReason::Version { daemon_version: 2, min: 2, max: 2 }, message: "no".into() },
1286            DaemonFrame::Refused { reason: RefuseReason::ReloadRefused { why: "older".into() }, message: "no".into() },
1287            DaemonFrame::Pong { pid: 1, uptime_s: 2, sessions: 0 },
1288            DaemonFrame::SessionList { sessions: vec![meta()] },
1289            DaemonFrame::Attached(AttachedWire::new(
1290                ClientId(1),
1291                AttachSnapshot {
1292                    meta: meta(),
1293                    view: view(),
1294                    conversation: conv(),
1295                    streaming: false,
1296                    replay: vec![],
1297                    pending_prompts: vec![],
1298                    clients: vec![],
1299                    input_owner: None,
1300                    display_tail: None,
1301                },
1302            )),
1303            DaemonFrame::Event(env(SessionEventWire::SystemNotice("x".into())).into()),
1304            DaemonFrame::Error { session_id: None, message: "e".into() },
1305            DaemonFrame::Bye { reason: None },
1306            DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation: 2, retry_after_ms: 500 }) },
1307            DaemonFrame::Bye { reason: Some(ByeReason::Shutdown) },
1308        ];
1309        for f in frames {
1310            let line = encode_line(&f).unwrap();
1311            let back: DaemonFrame = decode_line(line.trim_end()).unwrap();
1312            assert_eq!(serde_json::to_string(&back).unwrap(), line.trim_end());
1313        }
1314    }
1315
1316    #[test]
1317    fn hello_must_be_first_is_a_type_level_fact_and_version_is_exact() {
1318        let h = Hello::new(ClientKind::Test);
1319        assert_eq!(h.protocol_version, PROTOCOL_VERSION);
1320        assert_eq!(PROTOCOL_MIN, PROTOCOL_MAX);
1321        let json = serde_json::to_string(&ClientFrame::Hello(h)).unwrap();
1322        assert!(json.starts_with(r#"{"type":"hello""#));
1323        assert_eq!(PROTOCOL_VERSION, 3);
1324        // v1 frames (no `reconnect_of`, bare `bye`) still parse — the refusal is by version, not by shape.
1325        let h: Hello = serde_json::from_str(
1326            r#"{"protocol_version":1,"client":{"kind":"tui","terminal":null,"instance":"i"},"cwd":"/","client_version":"0"}"#,
1327        )
1328        .unwrap();
1329        assert!(h.reconnect_of.is_none());
1330        // Legacy Hello without env field → None (backwards compatible).
1331        assert!(h.env.is_none());
1332        // Hello::new captures env.
1333        let h2 = Hello::new(ClientKind::Test);
1334        assert!(h2.env.is_some(), "Hello::new must capture client env");
1335        let b: DaemonFrame = serde_json::from_str(r#"{"type":"bye"}"#).unwrap();
1336        assert!(matches!(b, DaemonFrame::Bye { reason: None }));
1337    }
1338
1339    #[test]
1340    fn debug_redacts_answer_and_submit_bodies() {
1341        let secret = "hunter2-very-secret";
1342        let f = ClientFrame::Cmd {
1343            session_id: "s".into(),
1344            cmd: SessionCommand::Answer { prompt_id: 1, value: Some(secret.into()) },
1345        };
1346        let d = format!("{f:?}");
1347        assert!(!d.contains(secret), "{d}");
1348        assert!(d.contains("redacted"));
1349        let f = ClientFrame::Cmd {
1350            session_id: "s".into(),
1351            cmd: SessionCommand::Submit { text: "my private prompt".into(), attachments: vec![] },
1352        };
1353        let d = format!("{f:?}");
1354        assert!(!d.contains("private"), "{d}");
1355        let f = ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Steer { text: "steer me".into() } };
1356        let d = format!("{f:?}");
1357        assert!(!d.contains("steer me"));
1358        // No lengths either: a char count is a side channel.
1359        assert!(!d.contains("chars"), "{d}");
1360        assert!(!d.contains('8'), "{d}");
1361    }
1362
1363    #[test]
1364    fn oversize_frame_rejected_both_directions() {
1365        assert!(DAEMON_MAX_FRAME_BYTES > RPC_MAX_FRAME_BYTES);
1366        let big = "x".repeat(DAEMON_MAX_FRAME_BYTES + 1);
1367        assert!(decode_line::<ClientFrame>(&big).unwrap_err().contains("64 MiB"));
1368        // encode side: a frame that would serialise past the cap is refused
1369        let f = DaemonFrame::Error { session_id: None, message: big };
1370        assert!(encode_line(&f).unwrap_err().contains("64 MiB"));
1371        // a > 1 MiB (rpc cap) frame is fine on the daemon wire
1372        let f = DaemonFrame::Error { session_id: None, message: "y".repeat(RPC_MAX_FRAME_BYTES * 2) };
1373        let line = encode_line(&f).unwrap();
1374        assert!(decode_line::<DaemonFrame>(line.trim_end()).is_ok());
1375    }
1376
1377    #[test]
1378    fn no_secret_bearing_fields_on_handshake_frames() {
1379        // Structural guard: the JSON of Welcome/Attached must not contain key-ish names.
1380        let w = serde_json::to_string(&Welcome {
1381            protocol_version: 1,
1382            daemon_version: "v".into(),
1383            pid: 1,
1384            profile: Some("p".into()),
1385            sessions: vec![meta()],
1386            progressive_tool_disclosure: false,
1387            generation: 1,
1388        })
1389        .unwrap()
1390        .to_lowercase();
1391        for bad in ["token", "secret", "api_key", "apikey", "password", "credential"] {
1392            assert!(!w.contains(bad), "{bad} in Welcome");
1393        }
1394    }
1395}
1396
1397#[cfg(test)]
1398mod float_roundtrip_tests {
1399    use super::*;
1400
1401    /// L≡S (phase 4 G9): the session cost crosses the wire as an f64. Without
1402    /// serde_json's `float_roundtrip` the parse is one ulp short for values
1403    /// like `0.000105 + 0.000045` and the footer rounds `$0.0002` → `$0.0001`.
1404    #[test]
1405    fn conversation_digest_cost_survives_the_wire_exactly() {
1406        let cost = (10f64 / 1_000_000.0) * 3.0 + (5f64 / 1_000_000.0) * 15.0
1407            + ((10f64 / 1_000_000.0) * 3.0 + (1f64 / 1_000_000.0) * 15.0);
1408        let snap = ConversationSnapshot { cost, ..Default::default() };
1409        let d = ConversationDigest::of(&snap);
1410        let line = encode_line(&d).unwrap();
1411        let back: ConversationDigest = decode_line(line.trim_end()).unwrap();
1412        assert_eq!(back.cost.to_bits(), cost.to_bits(), "{line}");
1413        assert_eq!(format!("{:.4}", back.cost), format!("{:.4}", cost));
1414    }
1415}
agent_engine/session/wire.rs

agent_engine/session/
wire.rs
