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: messages_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        }
432    }
433}
434
435// ── mirror of SessionEventWire (lossless except Conversation → digest) ────────
436
437#[derive(Debug, Clone, Serialize, Deserialize)]
438#[serde(tag = "ev", rename_all = "snake_case")]
439#[allow(clippy::large_enum_variant)] // mirrors SessionEventWire by design
440pub enum WireSessionEvent {
441    Stream { event: WireStreamEvent },
442    TurnStarted {
443        turn_baseline: usize,
444        trigger: TurnTrigger,
445        #[serde(default)]
446        user_text: Option<String>,
447    },
448    /// Digest only — see module docs.
449    Conversation { digest: ConversationDigest },
450    Prompt { request: PromptRequest },
451    PromptResolved { prompt_id: u64 },
452    External { event: crate::events::types::Event },
453    AutoTurnCapReached { cap: u32 },
454    Idle,
455    Steered { text: String, delivered: bool },
456    Dequeued { text: String },
457    SystemNotice { text: String },
458    LoaderProgress { event: WireLoaderEvent },
459    ExtensionNotification { extension_id: String, method: String, params: serde_json::Value },
460    SettingChanged { applied: SettingApplied },
461    QueryResult { id: u64, value: serde_json::Value },
462    Attached { client: ClientId, snapshot: AttachedWire },
463    ClientJoined { client: ClientId, kind: ClientKind },
464    ClientLeft { client: ClientId },
465    Ended { reason: EndReason },
466    Aborted { context_saved: bool },
467    Cleared { session_id: String },
468    CompactionStarted { source: String, disclosure: String },
469    CompactionApplied {
470        previous_session_id: String,
471        session_id: String,
472        chains_advanced: Vec<String>,
473        queued_restored: Option<String>,
474        msg_count: usize,
475    },
476    CompactionFailed { message: String, panicked: bool },
477    CompactionCancelled,
478    SubagentRows { rows: Vec<crate::runtime::subagent::SubagentDisplayRow> },
479    Resumed {
480        id: u64,
481        old_id: String,
482        new_id: String,
483        via: Option<String>,
484        clamp_notice: Option<String>,
485    },
486    InputOwnerChanged { from: Option<ClientId>, to: Option<ClientId>, reason: OwnerChangeReason },
487    Refused { client: ClientId, command: String, reason: String },
488    AttachRefused { message: String },
489    Lifecycle { lifecycle: SessionLifecycle },
490    Reloading { generation: u64, retry_after_ms: u64 },
491    /// (E-P0) Driver armed.
492    DriverArmed {
493        plugin_id: String,
494        run_id: String,
495        models: Vec<crate::extensions::session_driver::Selection>,
496        selection: crate::extensions::session_driver::Selection,
497        #[serde(default)]
498        deadline_ms: Option<u64>,
499        #[serde(default)]
500        notice: String,
501    },
502    /// (E-P0) Driver revoked.
503    DriverRevoked {
504        reason: String,
505        #[serde(default)]
506        undelivered_steering: Vec<String>,
507    },
508    /// (E-P0) Driver turn outcome.
509    DriverTurnOutcome {
510        outcome: crate::extensions::session_driver::Outcome,
511        selection: crate::extensions::session_driver::Selection,
512        #[serde(default)]
513        feedback: Option<String>,
514    },
515    /// (E-P7) Spend ceiling breached; turn cancelled and driver revoked.
516    CostCapReached {
517        scope: String,
518        cost: f64,
519        cap: f64,
520    },
521    /// Forward-compat: an additive variant from a newer daemon.
522    #[serde(other)]
523    Unknown,
524}
525
526#[derive(Debug, Clone, Serialize, Deserialize)]
527#[serde(tag = "kind", rename_all = "snake_case")]
528pub enum WireStreamEvent {
529    Llm(WireLlmEvent),
530    Session(WireSessionStreamEvent),
531    Agent(WireAgentEvent),
532}
533
534#[derive(Debug, Clone, Serialize, Deserialize)]
535#[serde(tag = "llm", rename_all = "snake_case")]
536pub enum WireLlmEvent {
537    Thinking { text: String },
538    Text { text: String },
539    ToolUseStart { tool_name: String, tool_id: String },
540    ToolUseDelta { tool_id: String, delta: String },
541    ToolUse { tool_name: String, tool_id: String, input: serde_json::Value },
542    ToolResult { tool_id: String, result: String },
543    ToolResultDelta { tool_id: String, delta: String },
544    ResponseStart,
545    ResponseReset,
546}
547
548#[derive(Debug, Clone, Serialize, Deserialize)]
549#[serde(tag = "session", rename_all = "snake_case")]
550pub enum WireSessionStreamEvent {
551    MessageHistory { messages: Vec<crate::SharedMessage> },
552    Usage {
553        input_tokens: u64,
554        output_tokens: u64,
555        cache_read_input_tokens: u64,
556        cache_creation_input_tokens: u64,
557        cache_creation_5m: Option<u64>,
558        cache_creation_1h: Option<u64>,
559        model: Option<String>,
560    },
561    Done,
562    Error { message: String, outcome: TurnOutcome },
563    Notice { text: String },
564}
565
566#[derive(Debug, Clone, Serialize, Deserialize)]
567#[serde(tag = "agent", rename_all = "snake_case")]
568pub enum WireAgentEvent {
569    SubagentStart { subagent_id: u64, agent_name: String, task_preview: String },
570    SubagentUpdate { subagent_id: u64, agent_name: String, status: String },
571    SubagentDone { subagent_id: u64, agent_name: String, result_preview: String, duration_secs: f64 },
572    SteeringDelivered { message: String },
573}
574
575#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
576pub struct WireLoadFailure {
577    pub plugin: String,
578    pub manifest_path: Option<PathBuf>,
579    pub reason: String,
580    pub hint: String,
581}
582
583#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
584#[serde(tag = "loader", rename_all = "snake_case")]
585pub enum WireLoaderEvent {
586    Started,
587    Loaded { plugin: String, loaded: usize, failed: usize },
588    Failed { failure: WireLoadFailure, loaded: usize, failed: usize },
589    Finished { loaded: Vec<String>, failed: Vec<WireLoadFailure> },
590}
591
592type LoadFailure = crate::extensions::manager::ExtensionLoadFailure;
593type LoaderEvent = crate::extensions::loader::ExtensionLoaderEvent;
594
595impl From<LoadFailure> for WireLoadFailure {
596    fn from(f: LoadFailure) -> Self {
597        Self { plugin: f.plugin, manifest_path: f.manifest_path, reason: f.reason, hint: f.hint }
598    }
599}
600impl From<WireLoadFailure> for LoadFailure {
601    fn from(f: WireLoadFailure) -> Self {
602        Self { plugin: f.plugin, manifest_path: f.manifest_path, reason: f.reason, hint: f.hint }
603    }
604}
605impl From<LoaderEvent> for WireLoaderEvent {
606    fn from(e: LoaderEvent) -> Self {
607        match e {
608            LoaderEvent::Started => Self::Started,
609            LoaderEvent::Loaded { plugin, loaded, failed } => Self::Loaded { plugin, loaded, failed },
610            LoaderEvent::Failed { failure, loaded, failed } => {
611                Self::Failed { failure: failure.into(), loaded, failed }
612            }
613            LoaderEvent::Finished { loaded, failed } => Self::Finished {
614                loaded,
615                failed: failed.into_iter().map(Into::into).collect(),
616            },
617        }
618    }
619}
620impl From<WireLoaderEvent> for LoaderEvent {
621    fn from(e: WireLoaderEvent) -> Self {
622        match e {
623            WireLoaderEvent::Started => Self::Started,
624            WireLoaderEvent::Loaded { plugin, loaded, failed } => Self::Loaded { plugin, loaded, failed },
625            WireLoaderEvent::Failed { failure, loaded, failed } => {
626                Self::Failed { failure: failure.into(), loaded, failed }
627            }
628            WireLoaderEvent::Finished { loaded, failed } => Self::Finished {
629                loaded,
630                failed: failed.into_iter().map(Into::into).collect(),
631            },
632        }
633    }
634}
635
636impl From<StreamEvent> for WireStreamEvent {
637    fn from(e: StreamEvent) -> Self {
638        match e {
639            StreamEvent::Llm(l) => Self::Llm(match l {
640                LlmEvent::Thinking(text) => WireLlmEvent::Thinking { text },
641                LlmEvent::Text(text) => WireLlmEvent::Text { text },
642                LlmEvent::ToolUseStart { tool_name, tool_id } => WireLlmEvent::ToolUseStart { tool_name, tool_id },
643                LlmEvent::ToolUseDelta { tool_id, delta } => WireLlmEvent::ToolUseDelta { tool_id, delta },
644                LlmEvent::ToolUse { tool_name, tool_id, input } => WireLlmEvent::ToolUse { tool_name, tool_id, input },
645                LlmEvent::ToolResult { tool_id, result } => WireLlmEvent::ToolResult { tool_id, result },
646                LlmEvent::ToolResultDelta { tool_id, delta } => WireLlmEvent::ToolResultDelta { tool_id, delta },
647                LlmEvent::ResponseStart => WireLlmEvent::ResponseStart,
648                LlmEvent::ResponseReset => WireLlmEvent::ResponseReset,
649            }),
650            StreamEvent::Session(s) => Self::Session(match s {
651                SessionEvent::MessageHistory(messages) => WireSessionStreamEvent::MessageHistory { messages },
652                SessionEvent::Usage {
653                    input_tokens,
654                    output_tokens,
655                    cache_read_input_tokens,
656                    cache_creation_input_tokens,
657                    cache_creation_5m,
658                    cache_creation_1h,
659                    model,
660                } => WireSessionStreamEvent::Usage {
661                    input_tokens,
662                    output_tokens,
663                    cache_read_input_tokens,
664                    cache_creation_input_tokens,
665                    cache_creation_5m,
666                    cache_creation_1h,
667                    model,
668                },
669                SessionEvent::Done => WireSessionStreamEvent::Done,
670                SessionEvent::Error(TurnError { message, outcome }) => WireSessionStreamEvent::Error { message, outcome },
671                SessionEvent::Notice(text) => WireSessionStreamEvent::Notice { text },
672                // merge(112): ContextHeadCheckpoint is actor-internal (Wall 1); never crosses the wire.
673                // Drop it silently — the actor handles persistence before forwarding.
674                SessionEvent::ContextHeadCheckpoint { .. } => WireSessionStreamEvent::Done,
675            }),
676            StreamEvent::Agent(a) => Self::Agent(match a {
677                AgentEvent::SubagentStart { subagent_id, agent_name, task_preview } => {
678                    WireAgentEvent::SubagentStart { subagent_id, agent_name, task_preview }
679                }
680                AgentEvent::SubagentUpdate { subagent_id, agent_name, status } => {
681                    WireAgentEvent::SubagentUpdate { subagent_id, agent_name, status }
682                }
683                AgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs } => {
684                    WireAgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs }
685                }
686                AgentEvent::SteeringDelivered { message } => WireAgentEvent::SteeringDelivered { message },
687            }),
688        }
689    }
690}
691
692impl From<WireStreamEvent> for StreamEvent {
693    fn from(e: WireStreamEvent) -> Self {
694        match e {
695            WireStreamEvent::Llm(l) => Self::Llm(match l {
696                WireLlmEvent::Thinking { text } => LlmEvent::Thinking(text),
697                WireLlmEvent::Text { text } => LlmEvent::Text(text),
698                WireLlmEvent::ToolUseStart { tool_name, tool_id } => LlmEvent::ToolUseStart { tool_name, tool_id },
699                WireLlmEvent::ToolUseDelta { tool_id, delta } => LlmEvent::ToolUseDelta { tool_id, delta },
700                WireLlmEvent::ToolUse { tool_name, tool_id, input } => LlmEvent::ToolUse { tool_name, tool_id, input },
701                WireLlmEvent::ToolResult { tool_id, result } => LlmEvent::ToolResult { tool_id, result },
702                WireLlmEvent::ToolResultDelta { tool_id, delta } => LlmEvent::ToolResultDelta { tool_id, delta },
703                WireLlmEvent::ResponseStart => LlmEvent::ResponseStart,
704                WireLlmEvent::ResponseReset => LlmEvent::ResponseReset,
705            }),
706            WireStreamEvent::Session(s) => Self::Session(match s {
707                WireSessionStreamEvent::MessageHistory { messages } => SessionEvent::MessageHistory(messages),
708                WireSessionStreamEvent::Usage {
709                    input_tokens,
710                    output_tokens,
711                    cache_read_input_tokens,
712                    cache_creation_input_tokens,
713                    cache_creation_5m,
714                    cache_creation_1h,
715                    model,
716                } => SessionEvent::Usage {
717                    input_tokens,
718                    output_tokens,
719                    cache_read_input_tokens,
720                    cache_creation_input_tokens,
721                    cache_creation_5m,
722                    cache_creation_1h,
723                    model,
724                },
725                WireSessionStreamEvent::Done => SessionEvent::Done,
726                WireSessionStreamEvent::Error { message, outcome } => SessionEvent::Error(TurnError { message, outcome }),
727                WireSessionStreamEvent::Notice { text } => SessionEvent::Notice(text),
728            }),
729            WireStreamEvent::Agent(a) => Self::Agent(match a {
730                WireAgentEvent::SubagentStart { subagent_id, agent_name, task_preview } => {
731                    AgentEvent::SubagentStart { subagent_id, agent_name, task_preview }
732                }
733                WireAgentEvent::SubagentUpdate { subagent_id, agent_name, status } => {
734                    AgentEvent::SubagentUpdate { subagent_id, agent_name, status }
735                }
736                WireAgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs } => {
737                    AgentEvent::SubagentDone { subagent_id, agent_name, result_preview, duration_secs }
738                }
739                WireAgentEvent::SteeringDelivered { message } => AgentEvent::SteeringDelivered { message },
740            }),
741        }
742    }
743}
744
745impl From<SessionEventWire> for WireSessionEvent {
746    fn from(e: SessionEventWire) -> Self {
747        use SessionEventWire as S;
748        match e {
749            S::Stream(ev) => Self::Stream { event: ev.into() },
750            S::TurnStarted { turn_baseline, trigger, user_text } => {
751                Self::TurnStarted { turn_baseline, trigger, user_text }
752            }
753            S::Conversation(snapshot) => Self::Conversation { digest: ConversationDigest::of(&snapshot) },
754            S::Prompt(request) => Self::Prompt { request },
755            S::PromptResolved { prompt_id } => Self::PromptResolved { prompt_id },
756            S::External(event) => Self::External { event },
757            S::AutoTurnCapReached { cap } => Self::AutoTurnCapReached { cap },
758            S::Idle => Self::Idle,
759            S::Steered { text, delivered } => Self::Steered { text, delivered },
760            S::Dequeued { text } => Self::Dequeued { text },
761            S::SystemNotice(text) => Self::SystemNotice { text },
762            S::LoaderProgress(ev) => Self::LoaderProgress { event: ev.into() },
763            S::ExtensionNotification { extension_id, method, params } => {
764                Self::ExtensionNotification { extension_id, method, params }
765            }
766            S::SettingChanged(applied) => Self::SettingChanged { applied },
767            S::QueryResult { id, value } => Self::QueryResult { id, value },
768            S::Attached { client, snapshot } => Self::Attached { client, snapshot: AttachedWire::new(client, snapshot) },
769            S::ClientJoined { client, kind } => Self::ClientJoined { client, kind },
770            S::ClientLeft { client } => Self::ClientLeft { client },
771            S::Ended { reason } => Self::Ended { reason },
772            S::Aborted { context_saved } => Self::Aborted { context_saved },
773            S::Cleared { session_id } => Self::Cleared { session_id },
774            S::CompactionStarted { source, disclosure } => Self::CompactionStarted { source, disclosure },
775            S::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count } => {
776                Self::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count }
777            }
778            S::CompactionFailed { message, panicked } => Self::CompactionFailed { message, panicked },
779            S::CompactionCancelled => Self::CompactionCancelled,
780            S::SubagentRows(rows) => Self::SubagentRows { rows },
781            S::Resumed { id, old_id, new_id, via, clamp_notice } => {
782                Self::Resumed { id, old_id, new_id, via, clamp_notice }
783            }
784            S::InputOwnerChanged { from, to, reason } => Self::InputOwnerChanged { from, to, reason },
785            S::Refused { client, command, reason } => Self::Refused { client, command, reason },
786            S::AttachRefused { message } => Self::AttachRefused { message },
787            S::Lifecycle(lifecycle) => Self::Lifecycle { lifecycle },
788            S::Reloading { generation, retry_after_ms } => Self::Reloading { generation, retry_after_ms },
789            S::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice } => {
790                Self::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice }
791            }
792            S::DriverRevoked { reason, undelivered_steering } => {
793                Self::DriverRevoked { reason, undelivered_steering }
794            }
795            S::DriverTurnOutcome { outcome, selection, feedback } => {
796                Self::DriverTurnOutcome { outcome, selection, feedback }
797            }
798            S::CostCapReached { scope, cost, cap } => {
799                Self::CostCapReached { scope, cost, cap }
800            }
801        }
802    }
803}
804
805impl From<WireSessionEvent> for SessionEventWire {
806    fn from(e: WireSessionEvent) -> Self {
807        use WireSessionEvent as W;
808        match e {
809            W::Stream { event } => Self::Stream(event.into()),
810            W::TurnStarted { turn_baseline, trigger, user_text } => {
811                Self::TurnStarted { turn_baseline, trigger, user_text }
812            }
813            // Messages are not on the wire; `SocketTransport` fills them from
814            // its mirror before handing the envelope up.
815            W::Conversation { digest } => Self::Conversation(digest.into_snapshot(Vec::new())),
816            W::Prompt { request } => Self::Prompt(request),
817            W::PromptResolved { prompt_id } => Self::PromptResolved { prompt_id },
818            W::External { event } => Self::External(event),
819            W::AutoTurnCapReached { cap } => Self::AutoTurnCapReached { cap },
820            W::Idle => Self::Idle,
821            W::Steered { text, delivered } => Self::Steered { text, delivered },
822            W::Dequeued { text } => Self::Dequeued { text },
823            W::SystemNotice { text } => Self::SystemNotice(text),
824            W::LoaderProgress { event } => Self::LoaderProgress(event.into()),
825            W::ExtensionNotification { extension_id, method, params } => {
826                Self::ExtensionNotification { extension_id, method, params }
827            }
828            W::SettingChanged { applied } => Self::SettingChanged(applied),
829            W::QueryResult { id, value } => Self::QueryResult { id, value },
830            W::Attached { snapshot, .. } => {
831                let (client, snapshot) = snapshot.into_snapshot();
832                Self::Attached { client, snapshot }
833            }
834            W::ClientJoined { client, kind } => Self::ClientJoined { client, kind },
835            W::ClientLeft { client } => Self::ClientLeft { client },
836            W::Ended { reason } => Self::Ended { reason },
837            W::Aborted { context_saved } => Self::Aborted { context_saved },
838            W::Cleared { session_id } => Self::Cleared { session_id },
839            W::CompactionStarted { source, disclosure } => Self::CompactionStarted { source, disclosure },
840            W::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count } => {
841                Self::CompactionApplied { previous_session_id, session_id, chains_advanced, queued_restored, msg_count }
842            }
843            W::CompactionFailed { message, panicked } => Self::CompactionFailed { message, panicked },
844            W::CompactionCancelled => Self::CompactionCancelled,
845            W::SubagentRows { rows } => Self::SubagentRows(rows),
846            W::Resumed { id, old_id, new_id, via, clamp_notice } => {
847                Self::Resumed { id, old_id, new_id, via, clamp_notice }
848            }
849            W::InputOwnerChanged { from, to, reason } => Self::InputOwnerChanged { from, to, reason },
850            W::Refused { client, command, reason } => Self::Refused { client, command, reason },
851            W::AttachRefused { message } => Self::AttachRefused { message },
852            W::Lifecycle { lifecycle } => Self::Lifecycle(lifecycle),
853            W::Reloading { generation, retry_after_ms } => Self::Reloading { generation, retry_after_ms },
854            W::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice } => {
855                Self::DriverArmed { plugin_id, run_id, models, selection, deadline_ms, notice }
856            }
857            W::DriverRevoked { reason, undelivered_steering } => {
858                Self::DriverRevoked { reason, undelivered_steering }
859            }
860            W::DriverTurnOutcome { outcome, selection, feedback } => {
861                Self::DriverTurnOutcome { outcome, selection, feedback }
862            }
863            W::CostCapReached { scope, cost, cap } => {
864                Self::CostCapReached { scope, cost, cap }
865            }
866            W::Unknown => Self::SystemNotice("unknown event from a newer daemon (ignored)".into()),
867        }
868    }
869}
870
871// ── framing helpers ───────────────────────────────────────────────────────────
872
873/// Human-readable form of the cap for error frames.
874pub fn frame_limit_msg() -> String {
875    format!("frame exceeds {} MiB limit", DAEMON_MAX_FRAME_BYTES / (1024 * 1024))
876}
877
878/// Encode one frame as a single line (no embedded newline: serde_json
879/// escapes them). Refuses to produce a line over [`DAEMON_MAX_FRAME_BYTES`].
880pub fn encode_line<T: Serialize>(frame: &T) -> Result<String, String> {
881    let mut s = serde_json::to_string(frame).map_err(|e| e.to_string())?;
882    if s.len() + 1 > DAEMON_MAX_FRAME_BYTES {
883        return Err(frame_limit_msg());
884    }
885    s.push('\n');
886    Ok(s)
887}
888
889/// Decode one line, enforcing [`DAEMON_MAX_FRAME_BYTES`].
890pub fn decode_line<'a, T: Deserialize<'a>>(line: &'a str) -> Result<T, String> {
891    if line.len() > DAEMON_MAX_FRAME_BYTES {
892        return Err(frame_limit_msg());
893    }
894    serde_json::from_str(line).map_err(|e| e.to_string())
895}
896
897#[cfg(test)]
898mod tests {
899    use super::*;
900    use crate::BudgetDimension;
901    use std::sync::Arc;
902
903    fn view() -> RuntimeView {
904        crate::session::handle::echo::view_for()
905    }
906
907    fn meta() -> SessionMeta {
908        crate::session::handle::echo::meta_for(&SessionId::from("wire-1"))
909    }
910
911    fn prompt() -> PromptRequest {
912        PromptRequest {
913            id: 3,
914            kind: PromptKind::Secret,
915            title: "sudo".into(),
916            prompt: "password".into(),
917            raised_at: chrono::Utc::now(),
918        }
919    }
920
921    fn conv() -> ConversationSnapshot {
922        ConversationSnapshot {
923            header: SessionHeader {
924                id: "wire-1".into(),
925                title: "t".into(),
926                name: None,
927                model: "m".into(),
928                thinking_level: "off".into(),
929                system_prompt: Some("sys".into()),
930                created_at: chrono::Utc::now(),
931                updated_at: chrono::Utc::now(),
932                parent_session: Some("p".into()),
933            },
934            api_messages: vec![Arc::new(serde_json::json!({"role":"user","content":"hi"}))],
935            messages_len: 1,
936            tokens: ConversationTokens { input: 1, output: 2, cache_read: 3, cache_creation: 4 },
937            cost: 0.5,
938            abort_context: Some("ctx".into()),
939            queued_message: None,
940            pending_events_len: 2,
941            consecutive_auto_turns: 1,
942        }
943    }
944
945    fn env(event: SessionEventWire) -> Envelope {
946        Envelope { session_id: SessionId::from("wire-1"), seq: 9, ts: chrono::Utc::now(), event }
947    }
948
949    fn stream_fixtures() -> Vec<StreamEvent> {
950        vec![
951            StreamEvent::Llm(LlmEvent::Thinking("t".into())),
952            StreamEvent::Llm(LlmEvent::Text("x\ny".into())),
953            StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name: "bash".into(), tool_id: "id1".into() }),
954            StreamEvent::Llm(LlmEvent::ToolUseDelta { tool_id: "id1".into(), delta: "{\"c".into() }),
955            StreamEvent::Llm(LlmEvent::ToolUse {
956                tool_name: "bash".into(),
957                tool_id: "id1".into(),
958                input: serde_json::json!({"command":"pwd"}),
959            }),
960            StreamEvent::Llm(LlmEvent::ToolResult { tool_id: "id1".into(), result: "/tmp".into() }),
961            StreamEvent::Llm(LlmEvent::ToolResultDelta { tool_id: "id1".into(), delta: "/t".into() }),
962            StreamEvent::Session(SessionEvent::MessageHistory(vec![Arc::new(serde_json::json!({"role":"user","content":"a"}))])),
963            StreamEvent::Session(SessionEvent::Usage {
964                input_tokens: 1,
965                output_tokens: 2,
966                cache_read_input_tokens: 3,
967                cache_creation_input_tokens: 4,
968                cache_creation_5m: Some(5),
969                cache_creation_1h: None,
970                model: Some("m".into()),
971            }),
972            StreamEvent::Session(SessionEvent::Done),
973            StreamEvent::Session(SessionEvent::Error(TurnError::provider("boom", "auth_error", "turn-1-0"))),
974            StreamEvent::Session(SessionEvent::Error(TurnError::budget(BudgetDimension::ToolCalls))),
975            StreamEvent::Session(SessionEvent::Error(TurnError { message: "c".into(), outcome: TurnOutcome::Canceled })),
976            StreamEvent::Session(SessionEvent::Notice("n".into())),
977            StreamEvent::Agent(AgentEvent::SubagentStart { subagent_id: 1, agent_name: "a".into(), task_preview: "p".into() }),
978            StreamEvent::Agent(AgentEvent::SubagentUpdate { subagent_id: 1, agent_name: "a".into(), status: "s".into() }),
979            StreamEvent::Agent(AgentEvent::SubagentDone {
980                subagent_id: 1,
981                agent_name: "a".into(),
982                result_preview: "r".into(),
983                duration_secs: 1.5,
984            }),
985            StreamEvent::Agent(AgentEvent::SteeringDelivered { message: "m".into() }),
986        ]
987    }
988
989    fn fixtures() -> Vec<SessionEventWire> {
990        use SessionEventWire as S;
991        let loader = crate::extensions::loader::ExtensionLoaderEvent::Finished {
992            loaded: vec!["p".into()],
993            failed: vec![LoadFailure {
994                plugin: "bad".into(),
995                manifest_path: Some(PathBuf::from("/x/manifest.json")),
996                reason: "oops".into(),
997                hint: "fix".into(),
998            }],
999        };
1000        let mut v: Vec<S> = stream_fixtures().into_iter().map(S::Stream).collect();
1001        v.extend([
1002            S::TurnStarted { turn_baseline: 4, trigger: TurnTrigger::EventAuto, user_text: None },
1003            S::TurnStarted { turn_baseline: 5, trigger: TurnTrigger::QueuedAuto, user_text: Some("q".into()) },
1004            S::TurnStarted { turn_baseline: 6, trigger: TurnTrigger::DriverAuto, user_text: None },
1005            S::Conversation(conv()),
1006            S::Prompt(prompt()),
1007            S::PromptResolved { prompt_id: 3 },
1008            S::External(crate::events::types::Event::simple("cli", "hello", None)),
1009            S::AutoTurnCapReached { cap: 5 },
1010            S::Idle,
1011            S::Steered { text: "s".into(), delivered: true },
1012            S::Dequeued { text: "d".into() },
1013            S::SystemNotice("sys".into()),
1014            S::LoaderProgress(loader),
1015            S::ExtensionNotification { extension_id: "e".into(), method: "widget.upsert".into(), params: serde_json::json!({"a":1}) },
1016            S::SettingChanged(SettingApplied {
1017                id: 3,
1018                setting: "model".into(),
1019                ok: true,
1020                message: None,
1021                view: view(),
1022                clamp: Some(ReasoningClampWire { from: "ultra".into(), to: "high".into() }),
1023            }),
1024            S::QueryResult { id: 1, value: serde_json::json!([1, 2]) },
1025            S::Attached {
1026                client: ClientId(2),
1027                snapshot: AttachSnapshot {
1028                    meta: meta(),
1029                    view: view(),
1030                    conversation: conv(),
1031                    streaming: true,
1032                    replay: vec![env(S::Stream(StreamEvent::Llm(LlmEvent::Text("partial".into()))))],
1033                    pending_prompts: vec![prompt()],
1034                    clients: vec![(ClientId(1), ClientKind::Tui)],
1035                    input_owner: Some(ClientId(1)),
1036                    display_tail: None,
1037                },
1038            },
1039            S::ClientJoined { client: ClientId(1), kind: ClientKind::Attach },
1040            S::ClientLeft { client: ClientId(1) },
1041            S::Ended { reason: EndReason::HostShutdown },
1042            S::Aborted { context_saved: true },
1043            S::Cleared { session_id: "new-1".into() },
1044            S::CompactionStarted { source: "manual".into(), disclosure: "disc".into() },
1045            S::CompactionApplied {
1046                previous_session_id: "old".into(),
1047                session_id: "new".into(),
1048                chains_advanced: vec!["c1".into()],
1049                queued_restored: Some("q".into()),
1050                msg_count: 12,
1051            },
1052            S::CompactionFailed { message: "boom".into(), panicked: false },
1053            S::CompactionCancelled,
1054            S::SubagentRows(vec![crate::runtime::subagent::SubagentDisplayRow {
1055                subagent_id: 1,
1056                agent_name: "a".into(),
1057                status: crate::runtime::subagent::SubagentStatus::Failed("x".into()),
1058                cancel_requested: false,
1059                elapsed_secs: 1.25,
1060                finished_elapsed: Some(std::time::Duration::from_millis(1250)),
1061            }]),
1062            S::Resumed { id: 2, old_id: "o".into(), new_id: "n".into(), via: Some("name".into()), clamp_notice: None },
1063            S::InputOwnerChanged { from: Some(ClientId(1)), to: Some(ClientId(2)), reason: OwnerChangeReason::Takeover },
1064            S::Refused { client: ClientId(2), command: "submit".into(), reason: "input owned by client #1".into() },
1065            S::AttachRefused { message: "parked".into() },
1066            S::Lifecycle(SessionLifecycle::Parked),
1067            S::Reloading { generation: 2, retry_after_ms: 500 },
1068            S::DriverArmed {
1069                plugin_id: "autonomous".into(),
1070                run_id: "run-1".into(),
1071                models: vec![
1072                    crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1073                    crate::extensions::session_driver::Selection { model: "m2".into(), effort: "low".into() },
1074                ],
1075                selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1076                deadline_ms: Some(60_000),
1077                notice: "armed".into(),
1078            },
1079            S::DriverRevoked { reason: "canceled".into(), undelivered_steering: vec!["steer1".into()] },
1080            S::DriverTurnOutcome {
1081                outcome: crate::extensions::session_driver::Outcome::Success,
1082                selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1083                feedback: Some("changed".into()),
1084            },
1085        ]);
1086        v
1087    }
1088
1089    #[test]
1090    fn wire_roundtrip_every_variant() {
1091        let all = fixtures();
1092        // 36 non-Stream variants (TurnStarted thrice = 3; +3 driver events) + 18 StreamEvent leaves.
1093        assert_eq!(all.len(), 36 + 18);
1094        for ev in all {
1095            let is_conv = matches!(ev, SessionEventWire::Conversation(_));
1096            let e = env(ev);
1097            let before = format!("{e:?}");
1098            let wire: WireEnvelope = e.into();
1099            let line = encode_line(&wire).unwrap();
1100            assert_eq!(line.matches('\n').count(), 1, "one line per frame");
1101            let back: WireEnvelope = decode_line(line.trim_end()).unwrap();
1102            let after = format!("{:?}", Envelope::from(back));
1103            if is_conv {
1104                // Digest on the wire: everything but the messages survives.
1105                assert!(!line.contains("api_messages"), "{line}");
1106                assert!(after.contains("api_messages: []"), "{after}");
1107                assert_eq!(before.replace("api_messages: [Object {\"content\": String(\"hi\"), \"role\": String(\"user\")}]", "api_messages: []"), after);
1108            } else {
1109                assert_eq!(before, after);
1110            }
1111        }
1112    }
1113
1114    #[test]
1115    fn hello_history_fields_default_and_roundtrip() {
1116        // Old peers omit the fields: Full / 120.
1117        let legacy = r#"{"kind":"tui","terminal":null,"instance":"i"}"#;
1118        let m: ClientMeta = serde_json::from_str(legacy).unwrap();
1119        assert_eq!(m.history, HistoryMode::Full);
1120        assert_eq!(m.tail_items, DEFAULT_TAIL_ITEMS);
1121        let h = Hello::new(ClientKind::Tui).with_history(HistoryMode::Digest).with_tail_items(9);
1122        let line = encode_line(&ClientFrame::Hello(h)).unwrap();
1123        assert!(line.contains("\"history\":\"digest\""), "{line}");
1124        assert!(line.contains("\"tail_items\":9"), "{line}");
1125        match decode_line::<ClientFrame>(line.trim_end()).unwrap() {
1126            ClientFrame::Hello(h) => {
1127                assert_eq!(h.client.history, HistoryMode::Digest);
1128                assert_eq!(h.client.tail_items, 9);
1129            }
1130            other => panic!("{other:?}"),
1131        }
1132        // Digest snapshots keep messages_len even with api_messages empty.
1133        let d = ConversationDigest::of(&conv());
1134        let snap = d.into_snapshot(Vec::new());
1135        assert_eq!(snap.messages_len, 1);
1136        assert!(snap.api_messages.is_empty());
1137        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();
1138        assert_eq!(legacy.messages_len, 0);
1139    }
1140
1141    #[test]
1142    fn conversation_digest_roundtrips_and_detects_drift() {
1143        let c = conv();
1144        let d = ConversationDigest::of(&c);
1145        assert_eq!(d.messages_len, 1);
1146        assert!(d.matches(&c.api_messages));
1147        assert!(!d.matches(&[]));
1148        let mut other = c.api_messages.clone();
1149        other.push(Arc::new(serde_json::json!({"role":"assistant","content":"x"})));
1150        assert!(!d.matches(&other));
1151        let back = d.clone().into_snapshot(c.api_messages.clone());
1152        assert_eq!(ConversationDigest::of(&back), d);
1153        // stable across runs/binaries: FNV-1a offset basis for the empty list
1154        assert_eq!(messages_hash(&[]), 0xcbf2_9ce4_8422_2325);
1155        assert_ne!(messages_hash(&c.api_messages), messages_hash(&[]));
1156    }
1157
1158    #[test]
1159    fn unknown_event_variant_tolerated() {
1160        let v: WireSessionEvent = serde_json::from_str(r#"{"ev":"from_the_future","x":1}"#).unwrap();
1161        assert!(matches!(v, WireSessionEvent::Unknown));
1162        assert!(matches!(SessionEventWire::from(v), SessionEventWire::SystemNotice(_)));
1163    }
1164
1165    #[test]
1166    fn driver_armed_round_trips_and_old_client_skips() {
1167        // A v3 daemon emits DriverArmed; a v2 client decodes it as Unknown.
1168        let armed = WireSessionEvent::DriverArmed {
1169            plugin_id: "autonomous".into(),
1170            run_id: "run-42".into(),
1171            models: vec![
1172                crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1173            ],
1174            selection: crate::extensions::session_driver::Selection { model: "m1".into(), effort: "high".into() },
1175            deadline_ms: Some(120_000),
1176            notice: "armed and ready".into(),
1177        };
1178        let json = serde_json::to_string(&armed).unwrap();
1179        // v3 client: exact round-trip.
1180        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
1181        assert!(matches!(&back, WireSessionEvent::DriverArmed { plugin_id, .. } if plugin_id == "autonomous"));
1182        // Simulated v2 client: rename tag so it hits Unknown.
1183        let old_json = json.replace(r#""ev":"driver_armed""#, r#""ev":"v2_unknown_event""#);
1184        let old: WireSessionEvent = serde_json::from_str(&old_json).unwrap();
1185        assert!(matches!(old, WireSessionEvent::Unknown));
1186    }
1187
1188    #[test]
1189    fn driver_revoked_and_outcome_round_trip() {
1190        let revoked = WireSessionEvent::DriverRevoked {
1191            reason: "canceled".into(),
1192            undelivered_steering: vec!["steer1".into(), "steer2".into()],
1193        };
1194        let json = serde_json::to_string(&revoked).unwrap();
1195        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
1196        assert!(matches!(&back, WireSessionEvent::DriverRevoked { reason, undelivered_steering }
1197            if reason == "canceled" && undelivered_steering.len() == 2));
1198
1199        let outcome = WireSessionEvent::DriverTurnOutcome {
1200            outcome: crate::extensions::session_driver::Outcome::Success,
1201            selection: crate::extensions::session_driver::Selection { model: "m".into(), effort: "high".into() },
1202            feedback: Some("changed".into()),
1203        };
1204        let json = serde_json::to_string(&outcome).unwrap();
1205        let back: WireSessionEvent = serde_json::from_str(&json).unwrap();
1206        assert!(matches!(&back, WireSessionEvent::DriverTurnOutcome { feedback: Some(f), .. } if f == "changed"));
1207    }
1208
1209    #[test]
1210    fn driver_revoked_defaults_tolerate_missing_fields() {
1211        // Old daemon or minimal frame: undelivered_steering missing → empty vec.
1212        let minimal = r#"{"ev":"driver_revoked","reason":"timeout"}"#;
1213        let v: WireSessionEvent = serde_json::from_str(minimal).unwrap();
1214        assert!(matches!(&v, WireSessionEvent::DriverRevoked { reason, undelivered_steering }
1215            if reason == "timeout" && undelivered_steering.is_empty()));
1216    }
1217
1218    #[test]
1219    fn client_frames_roundtrip_and_daemon_frames_roundtrip() {
1220        let frames = vec![
1221            ClientFrame::Hello(Hello::new(ClientKind::Attach)),
1222            ClientFrame::Hello(Hello::new(ClientKind::Tui).with_history(HistoryMode::Digest).with_tail_items(7)),
1223            ClientFrame::Ping,
1224            ClientFrame::Sessions,
1225            ClientFrame::Shutdown { force: true },
1226            ClientFrame::Attach(Attach::Existing { session_id: "s".into(), mode: AttachMode::Mirror }),
1227            ClientFrame::Attach(Attach::Create { config: SessionConfig::default(), mode: AttachMode::Takeover }),
1228            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Submit { text: "hi".into(), attachments: vec![] } },
1229            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Answer { prompt_id: 1, value: Some("pw".into()) } },
1230            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::EngineCommand { id: 4, name: "model".into(), arg: "x".into() } },
1231            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Query { id: 5, query: SessionQuery::Status } },
1232            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Set { id: 6, setting: SessionSetting::ReloadPrompt } },
1233            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Checkpoint { reason: CheckpointReason::Reload } },
1234            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::KeepWarm { on: true } },
1235            ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::DriverStart {
1236                plugin: "autonomous".into(), command: "auto".into(), arg: "start -- do stuff".into(),
1237            }},
1238            ClientFrame::Reload { now: true, drain_secs: Some(3), exe: Some(PathBuf::from("/bin/synaps")) },
1239            ClientFrame::Purge,
1240            ClientFrame::Bye,
1241        ];
1242        for f in frames {
1243            let line = encode_line(&f).unwrap();
1244            let back: ClientFrame = decode_line(line.trim_end()).unwrap();
1245            assert_eq!(serde_json::to_string(&back).unwrap(), line.trim_end());
1246        }
1247        let frames = vec![
1248            DaemonFrame::Welcome(Welcome {
1249                protocol_version: PROTOCOL_VERSION,
1250                daemon_version: "0.9.0".into(),
1251                pid: 1,
1252                profile: None,
1253                sessions: vec![meta()],
1254                progressive_tool_disclosure: true,
1255                generation: 1,
1256            }),
1257            DaemonFrame::Refused { reason: RefuseReason::Version { daemon_version: 2, min: 2, max: 2 }, message: "no".into() },
1258            DaemonFrame::Refused { reason: RefuseReason::ReloadRefused { why: "older".into() }, message: "no".into() },
1259            DaemonFrame::Pong { pid: 1, uptime_s: 2, sessions: 0 },
1260            DaemonFrame::SessionList { sessions: vec![meta()] },
1261            DaemonFrame::Attached(AttachedWire::new(
1262                ClientId(1),
1263                AttachSnapshot {
1264                    meta: meta(),
1265                    view: view(),
1266                    conversation: conv(),
1267                    streaming: false,
1268                    replay: vec![],
1269                    pending_prompts: vec![],
1270                    clients: vec![],
1271                    input_owner: None,
1272                    display_tail: None,
1273                },
1274            )),
1275            DaemonFrame::Event(env(SessionEventWire::SystemNotice("x".into())).into()),
1276            DaemonFrame::Error { session_id: None, message: "e".into() },
1277            DaemonFrame::Bye { reason: None },
1278            DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation: 2, retry_after_ms: 500 }) },
1279            DaemonFrame::Bye { reason: Some(ByeReason::Shutdown) },
1280        ];
1281        for f in frames {
1282            let line = encode_line(&f).unwrap();
1283            let back: DaemonFrame = decode_line(line.trim_end()).unwrap();
1284            assert_eq!(serde_json::to_string(&back).unwrap(), line.trim_end());
1285        }
1286    }
1287
1288    #[test]
1289    fn hello_must_be_first_is_a_type_level_fact_and_version_is_exact() {
1290        let h = Hello::new(ClientKind::Test);
1291        assert_eq!(h.protocol_version, PROTOCOL_VERSION);
1292        assert_eq!(PROTOCOL_MIN, PROTOCOL_MAX);
1293        let json = serde_json::to_string(&ClientFrame::Hello(h)).unwrap();
1294        assert!(json.starts_with(r#"{"type":"hello""#));
1295        assert_eq!(PROTOCOL_VERSION, 3);
1296        // v1 frames (no `reconnect_of`, bare `bye`) still parse — the refusal is by version, not by shape.
1297        let h: Hello = serde_json::from_str(
1298            r#"{"protocol_version":1,"client":{"kind":"tui","terminal":null,"instance":"i"},"cwd":"/","client_version":"0"}"#,
1299        )
1300        .unwrap();
1301        assert!(h.reconnect_of.is_none());
1302        // Legacy Hello without env field → None (backwards compatible).
1303        assert!(h.env.is_none());
1304        // Hello::new captures env.
1305        let h2 = Hello::new(ClientKind::Test);
1306        assert!(h2.env.is_some(), "Hello::new must capture client env");
1307        let b: DaemonFrame = serde_json::from_str(r#"{"type":"bye"}"#).unwrap();
1308        assert!(matches!(b, DaemonFrame::Bye { reason: None }));
1309    }
1310
1311    #[test]
1312    fn debug_redacts_answer_and_submit_bodies() {
1313        let secret = "hunter2-very-secret";
1314        let f = ClientFrame::Cmd {
1315            session_id: "s".into(),
1316            cmd: SessionCommand::Answer { prompt_id: 1, value: Some(secret.into()) },
1317        };
1318        let d = format!("{f:?}");
1319        assert!(!d.contains(secret), "{d}");
1320        assert!(d.contains("redacted"));
1321        let f = ClientFrame::Cmd {
1322            session_id: "s".into(),
1323            cmd: SessionCommand::Submit { text: "my private prompt".into(), attachments: vec![] },
1324        };
1325        let d = format!("{f:?}");
1326        assert!(!d.contains("private"), "{d}");
1327        let f = ClientFrame::Cmd { session_id: "s".into(), cmd: SessionCommand::Steer { text: "steer me".into() } };
1328        let d = format!("{f:?}");
1329        assert!(!d.contains("steer me"));
1330        // No lengths either: a char count is a side channel.
1331        assert!(!d.contains("chars"), "{d}");
1332        assert!(!d.contains('8'), "{d}");
1333    }
1334
1335    #[test]
1336    fn oversize_frame_rejected_both_directions() {
1337        assert!(DAEMON_MAX_FRAME_BYTES > RPC_MAX_FRAME_BYTES);
1338        let big = "x".repeat(DAEMON_MAX_FRAME_BYTES + 1);
1339        assert!(decode_line::<ClientFrame>(&big).unwrap_err().contains("64 MiB"));
1340        // encode side: a frame that would serialise past the cap is refused
1341        let f = DaemonFrame::Error { session_id: None, message: big };
1342        assert!(encode_line(&f).unwrap_err().contains("64 MiB"));
1343        // a > 1 MiB (rpc cap) frame is fine on the daemon wire
1344        let f = DaemonFrame::Error { session_id: None, message: "y".repeat(RPC_MAX_FRAME_BYTES * 2) };
1345        let line = encode_line(&f).unwrap();
1346        assert!(decode_line::<DaemonFrame>(line.trim_end()).is_ok());
1347    }
1348
1349    #[test]
1350    fn no_secret_bearing_fields_on_handshake_frames() {
1351        // Structural guard: the JSON of Welcome/Attached must not contain key-ish names.
1352        let w = serde_json::to_string(&Welcome {
1353            protocol_version: 1,
1354            daemon_version: "v".into(),
1355            pid: 1,
1356            profile: Some("p".into()),
1357            sessions: vec![meta()],
1358            progressive_tool_disclosure: false,
1359            generation: 1,
1360        })
1361        .unwrap()
1362        .to_lowercase();
1363        for bad in ["token", "secret", "api_key", "apikey", "password", "credential"] {
1364            assert!(!w.contains(bad), "{bad} in Welcome");
1365        }
1366    }
1367}
1368
1369#[cfg(test)]
1370mod float_roundtrip_tests {
1371    use super::*;
1372
1373    /// L≡S (phase 4 G9): the session cost crosses the wire as an f64. Without
1374    /// serde_json's `float_roundtrip` the parse is one ulp short for values
1375    /// like `0.000105 + 0.000045` and the footer rounds `$0.0002` → `$0.0001`.
1376    #[test]
1377    fn conversation_digest_cost_survives_the_wire_exactly() {
1378        let cost = (10f64 / 1_000_000.0) * 3.0 + (5f64 / 1_000_000.0) * 15.0
1379            + ((10f64 / 1_000_000.0) * 3.0 + (1f64 / 1_000_000.0) * 15.0);
1380        let snap = ConversationSnapshot { cost, ..Default::default() };
1381        let d = ConversationDigest::of(&snap);
1382        let line = encode_line(&d).unwrap();
1383        let back: ConversationDigest = decode_line(line.trim_end()).unwrap();
1384        assert_eq!(back.cost.to_bits(), cost.to_bits(), "{line}");
1385        assert_eq!(format!("{:.4}", back.cost), format!("{:.4}", cost));
1386    }
1387}
agent_engine/session/wire.rs

agent_engine/session/
wire.rs
