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

agent_engine/session/
wire.rs
