agent_engine/daemon/
conn.rs

1//! Per-connection pump: handshake → control loop → attach → pump.
2//!
3//! Tracing here never logs frame bodies (`ClientFrame`'s `Debug` redacts
4//! `Answer`/`Submit`/`Steer`); `Answer` values are forwarded to the actor
5//! and nowhere else.
6
7use std::sync::atomic::Ordering;
8use std::sync::Arc;
9use std::time::Duration;
10
11use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
12use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
13use tokio::net::UnixStream;
14use tokio::sync::{broadcast, mpsc};
15use tokio_util::sync::CancellationToken;
16
17use super::DaemonState;
18use crate::session::handle::CMD_CHAN_CAP;
19use crate::session::transport::{TransportError, ATTACH_TIMEOUT, ATTACH_TIMEOUT_PARKED};
20use crate::session::wire::*;
21use crate::session::*;
22use crate::{SessionEvent, StreamEvent};
23
24const HELLO_TIMEOUT: Duration = Duration::from_secs(2);
25
26type Reader = BufReader<OwnedReadHalf>;
27
28/// Which `SessionCommand`s a socket client may originate (§4 MED). The UDS
29/// is 0600 same-uid, so this is a footgun guard, not an auth boundary:
30/// anything that drives the actor's *own* bookkeeping (`Attach`, `Detach`
31/// for another client, `Resync`, `HostEvent`) or ends the session for every
32/// other client with a host reason (`End`) is refused with an `Error` frame.
33/// The owner ends a session with `/close` (an engine command, checked by the
34/// actor); `daemon close` does it from a control connection
35/// (`ClientFrame::Close`), `daemon stop` for every session.
36fn client_may_send(cmd: &SessionCommand, own: ClientId) -> Result<(), &'static str> {
37    use SessionCommand as C;
38    match cmd {
39        C::Submit { .. }
40        | C::Steer { .. }
41        | C::SubagentSteer { .. }
42        | C::Cancel
43        | C::Answer { .. }
44        | C::Set { .. }
45        | C::Compact { .. }
46        | C::NewSession
47        | C::Save
48        | C::Query { .. }
49        | C::EngineCommand { .. }
50        | C::SubmitPrepared { .. }
51        | C::PluginCommand { .. }
52        | C::Resume { .. }
53        | C::KeepWarm { .. }
54        // `/takeover`: the actor checks the attach mode.
55        | C::TakeInput => Ok(()),
56        // A reload checkpoint stops every worker and fences the session
57        // for the exec that follows: only the daemon's own reload sends it.
58        C::Checkpoint { reason: crate::session::CheckpointReason::Reload } => {
59            Err("checkpoint(reload): host-only (use `synaps daemon reload`)")
60        }
61        C::Checkpoint { .. } => Ok(()),
62        C::Detach { client } if *client == own => Ok(()),
63        C::Detach { .. } => Err("detach: not your client id"),
64        C::Attach { .. } => Err("attach: already attached (one session per connection)"),
65        // C4: an owner may end its session (`/end`); the actor enforces
66        // ownership (B1 `is_input_command`), the conn only screens reasons.
67        C::End { reason: EndReason::ClientQuit } => Ok(()),
68        C::End { .. } => Err("end: only ClientQuit may come from a client (host reasons are the daemon's)"),
69        C::Resync { .. } => Err("resync: not a client command"),
70        C::Park => Err("park: not a client command"),
71        C::AdoptWorkers { .. } => Err("adopt_workers: not a client command"),
72        C::RestoreToolActivations { .. } => Err("restore_tool_activations: not a client command"),
73        C::ReloadAborted => Err("reload_aborted: not a client command"),
74        C::ArmPark => Err("arm_park: not a client command"),
75        C::HostEvent(_) => Err("host_event: not a client command"),
76        C::DriverStart { .. } => Ok(()),
77    }
78}
79
80/// Refused at the connection while a reload is in progress — from the drain
81/// on, before the session is even checkpointed. `SessionCommand::
82/// refused_while_reloading` minus what the drain must still allow: a
83/// `Steer`, `Set` or engine command for the turn the drain is waiting to
84/// finish. The actor's reload fence refuses those once it has checkpointed
85/// (it alone knows whether a `Steer` would start a turn).
86fn refused_while_draining(cmd: &SessionCommand) -> bool {
87    use SessionCommand as C;
88    cmd.refused_while_reloading()
89        && !matches!(
90            cmd,
91            C::Steer { .. } | C::SubagentSteer { .. } | C::Set { .. } | C::EngineCommand { .. } | C::KeepWarm { .. }
92        )
93}
94
95// One per line read, consumed at once: not worth boxing every frame.
96#[allow(clippy::large_enum_variant)]
97enum Read {
98    Frame(ClientFrame),
99    Eof,
100    Oversize,
101    Bad(String),
102}
103
104async fn read_frame(reader: &mut Reader, line: &mut String) -> std::io::Result<Read> {
105    loop {
106        line.clear();
107        let n = tokio::io::AsyncReadExt::take(&mut *reader, MAX_FRAME_BYTES as u64 + 1)
108            .read_line(line)
109            .await?;
110        if n == 0 {
111            return Ok(Read::Eof);
112        }
113        if n > MAX_FRAME_BYTES {
114            return Ok(Read::Oversize);
115        }
116        let t = line.trim_end();
117        if t.is_empty() {
118            continue;
119        }
120        return Ok(match decode_line::<ClientFrame>(t) {
121            Ok(f) => Read::Frame(f),
122            Err(e) => Read::Bad(e),
123        });
124    }
125}
126
127/// Writer task: owns the write half; bounded queue = per-client backpressure.
128fn spawn_writer(mut w: OwnedWriteHalf) -> (mpsc::Sender<DaemonFrame>, tokio::task::JoinHandle<()>) {
129    let (tx, mut rx) = mpsc::channel::<DaemonFrame>(CMD_CHAN_CAP);
130    let task = tokio::spawn(async move {
131        while let Some(frame) = rx.recv().await {
132            let bye = matches!(frame, DaemonFrame::Bye { .. } | DaemonFrame::Refused { .. });
133            match encode_line(&frame) {
134                Ok(line) => {
135                    if w.write_all(line.as_bytes()).await.is_err() {
136                        break;
137                    }
138                }
139                Err(e) => {
140                    // Symmetric cap: an outbound frame we cannot legally send
141                    // becomes an Error frame and the connection closes.
142                    tracing::warn!("daemon: encode failed: {e}");
143                    let err = DaemonFrame::Error { session_id: None, message: format!("daemon could not encode a frame: {e}") };
144                    if let Ok(line) = encode_line(&err) {
145                        let _ = w.write_all(line.as_bytes()).await;
146                    }
147                    let _ = w.shutdown().await;
148                    break;
149                }
150            }
151            if bye {
152                let _ = w.shutdown().await;
153                break;
154            }
155        }
156    });
157    (tx, task)
158}
159
160pub async fn serve(state: Arc<DaemonState>, stream: UnixStream, shutdown: CancellationToken) {
161    let (r, w) = stream.into_split();
162    let mut reader = BufReader::new(r);
163    let (tx, writer) = spawn_writer(w);
164    let mut line = String::new();
165
166    // ── handshake: first frame MUST be Hello ──
167    let hello = match tokio::time::timeout(HELLO_TIMEOUT, read_frame(&mut reader, &mut line)).await {
168        Ok(Ok(Read::Frame(ClientFrame::Hello(h)))) => h,
169        Ok(Ok(Read::Frame(_))) | Ok(Ok(Read::Bad(_))) => {
170            let _ = tx
171                .send(DaemonFrame::Refused { reason: RefuseReason::Protocol, message: "first frame must be hello".into() })
172                .await;
173            let _ = writer.await;
174            return;
175        }
176        Ok(Ok(Read::Oversize)) => {
177            let _ = tx.send(DaemonFrame::Error { session_id: None, message: frame_limit_msg() }).await;
178            drop(tx);
179            let _ = writer.await;
180            return;
181        }
182        _ => {
183            drop(tx);
184            let _ = writer.await;
185            return;
186        }
187    };
188    if hello.protocol_version < PROTOCOL_MIN || hello.protocol_version > PROTOCOL_MAX {
189        tracing::warn!(client = hello.protocol_version, daemon = PROTOCOL_VERSION, "daemon: refusing protocol version");
190        let _ = tx
191            .send(DaemonFrame::Refused {
192                reason: RefuseReason::Version { daemon_version: PROTOCOL_VERSION, min: PROTOCOL_MIN, max: PROTOCOL_MAX },
193                message: format!(
194                    "protocol version {} not supported (daemon speaks {}; daemon binary {})",
195                    hello.protocol_version,
196                    PROTOCOL_VERSION,
197                    binary_version()
198                ),
199            })
200            .await;
201        let _ = writer.await;
202        return;
203    }
204    // Clients validate their name before connecting; this is the backstop.
205    if let Some(Err(message)) = hello.client.name.as_deref().map(crate::session::validate_client_name) {
206        let _ = tx.send(DaemonFrame::Refused { reason: RefuseReason::Protocol, message }).await;
207        let _ = writer.await;
208        return;
209    }
210    let daemon_version = binary_version();
211    if hello.client_version != daemon_version {
212        tracing::info!(client = %hello.client_version, daemon = %daemon_version, "daemon: client binary version differs (same protocol; allowed)");
213    }
214    let welcome = Welcome {
215        protocol_version: PROTOCOL_VERSION,
216        daemon_version,
217        pid: std::process::id(),
218        profile: state.profile.clone(),
219        sessions: state.session_metas(),
220        progressive_tool_disclosure: state.host.parts().progressive_tool_disclosure,
221        generation: state.generation,
222        daemon_build: crate::session::wire::build_id().to_string(),
223    };
224    if tx.send(DaemonFrame::Welcome(welcome)).await.is_err() {
225        return;
226    }
227    tracing::debug!(kind = ?hello.client.kind, "daemon: client connected");
228
229    // ── control loop (no session allocated) ──
230    let attach = loop {
231        let announce = state.announce();
232        let frame = tokio::select! {
233            _ = shutdown.cancelled() => { let _ = tx.send(DaemonFrame::Bye { reason: None }).await; let _ = writer.await; return; }
234            _ = announce.cancelled() => {
235                let _ = tx.send(DaemonFrame::Bye { reason: Some(reload_bye(&state)) }).await;
236                let _ = writer.await;
237                return;
238            }
239            r = read_frame(&mut reader, &mut line) => r,
240        };
241        match frame {
242            Ok(Read::Frame(ClientFrame::Ping)) => {
243                // `daemon status` and `daemon reload` read daemon.json after a Ping.
244                state.heal_daemon_json();
245                let _ = tx
246                    .send(DaemonFrame::Pong { pid: std::process::id(), uptime_s: state.uptime_s(), sessions: state.live_sessions().len() })
247                    .await;
248            }
249            Ok(Read::Frame(ClientFrame::Sessions)) => {
250                let _ = tx.send(DaemonFrame::SessionList { sessions: state.session_metas() }).await;
251            }
252            Ok(Read::Frame(ClientFrame::Shutdown { force })) => {
253                tracing::info!(force, "daemon: shutdown requested over the socket");
254                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
255                let _ = writer.await;
256                state.request_shutdown(force);
257                return;
258            }
259            Ok(Read::Frame(ClientFrame::Attach(a))) => {
260                if matches!(a, Attach::Create { .. }) && state.reloading.load(Ordering::SeqCst) {
261                    let _ = tx
262                        .send(DaemonFrame::Refused { reason: RefuseReason::Busy, message: "daemon reloading; retry in a moment".into() })
263                        .await;
264                    continue;
265                }
266                break a;
267            }
268            Ok(Read::Frame(ClientFrame::Cmd { session_id, .. })) => {
269                let _ = tx.send(DaemonFrame::Error { session_id: Some(session_id), message: "not attached".into() }).await;
270            }
271            // C3: the whole §2.8 sequence runs on this connection's task;
272            // on success the process is replaced and this never returns.
273            Ok(Read::Frame(ref f @ ClientFrame::Reload { .. })) => {
274                let req = super::reload::ReloadRequest::from_frame(f).expect("reload frame");
275                match super::reload::prepare(&state, &state.paths, req).await {
276                    Err(super::reload::ReloadError::Refused(why)) => {
277                        let _ = tx
278                            .send(DaemonFrame::Refused { reason: RefuseReason::ReloadRefused { why: why.clone() }, message: why })
279                            .await;
280                    }
281                    Err(super::reload::ReloadError::ExecFailed(e)) => {
282                        let _ = tx.send(DaemonFrame::Error { session_id: None, message: format!("reload: {e}") }).await;
283                    }
284                    Ok(prepared) => {
285                        // Tell the requester, flush, close — then exec.
286                        let _ = tx.send(DaemonFrame::Bye { reason: Some(reload_bye(&state)) }).await;
287                        drop(tx);
288                        let _ = tokio::time::timeout(
289                            Duration::from_secs(crate::session::budgets::RELOAD_BYE_FLUSH_SECS),
290                            writer,
291                        )
292                        .await;
293                        let e = super::reload::exec(&state, prepared).await;
294                        tracing::error!(error = %e, "daemon: reload did not exec");
295                        return;
296                    }
297                }
298            }
299            // `daemon close`: end one session with host authority. A
300            // refusal is terminal (the writer closes after `Refused`).
301            Ok(Read::Frame(ClientFrame::Close { session_id, force, attached })) => {
302                tracing::info!(session = %session_id, force, attached, "daemon: close requested over the socket");
303                let reply = super::close::close_session(&state, &session_id, force, attached).await;
304                let _ = tx.send(reply).await;
305            }
306            // `daemon pin`: pin/unpin one session with host authority. A
307            // refusal is terminal, like `Close`.
308            Ok(Read::Frame(ClientFrame::Pin { session_id, on })) => {
309                tracing::info!(session = %session_id, on, "daemon: pin requested over the socket");
310                let reply = super::pin::pin_session(&state, &session_id, on).await;
311                let _ = tx.send(reply).await;
312            }
313            // `daemon reload-extension`: restart one extension in place,
314            // sessions untouched. A refusal is terminal, like `Close`.
315            Ok(Read::Frame(ClientFrame::ReloadExtension { extension })) => {
316                tracing::info!(extension = %extension, "daemon: extension reload requested over the socket");
317                let reply = super::extension_reload::reload_extension(&state, &extension).await;
318                let _ = tx.send(reply).await;
319            }
320            // C2: jemalloc purge in the daemon (bench hygiene); reply Pong.
321            Ok(Read::Frame(ClientFrame::Purge)) => {
322                agent_core::core::memstat::purge_arenas();
323                let _ = tx
324                    .send(DaemonFrame::Pong { pid: std::process::id(), uptime_s: state.uptime_s(), sessions: state.live_sessions().len() })
325                    .await;
326            }
327            // `synaps auth state`: the host broker's in-memory account state.
328            Ok(Read::Frame(ClientFrame::AuthState(query))) => {
329                let _ = tx.send(auth_state_reply(&state, &query)).await;
330            }
331            Ok(Read::Frame(ClientFrame::Hello(_))) => {
332                let _ = tx.send(DaemonFrame::Error { session_id: None, message: "duplicate hello".into() }).await;
333            }
334            Ok(Read::Frame(ClientFrame::Bye)) | Ok(Read::Eof) | Err(_) => {
335                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
336                let _ = writer.await;
337                return;
338            }
339            Ok(Read::Oversize) => {
340                let _ = tx.send(DaemonFrame::Error { session_id: None, message: frame_limit_msg() }).await;
341                drop(tx);
342                let _ = writer.await;
343                return;
344            }
345            Ok(Read::Bad(e)) => {
346                let _ = tx.send(DaemonFrame::Error { session_id: None, message: format!("malformed frame: {e}") }).await;
347            }
348        }
349    };
350
351    // ── attach ──
352    let mut attached_to_live = false;
353    let (handle, mode) = match attach {
354        Attach::Existing { session_id, mode } => match state.attach(&session_id) {
355            Some(h) if h.meta().locked_by.is_some() => (state.retry_locked_placeholder(h).await, mode),
356            Some(h) => (h, mode),
357            None => {
358                let _ = tx
359                    .send(DaemonFrame::Error { session_id: Some(session_id), message: "unknown session".into() })
360                    .await;
361                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
362                let _ = writer.await;
363                return;
364            }
365        },
366        Attach::Create { mut config, mode } => {
367            // Recorded by the actor from `/memory` (and carried by reload's
368            // rehydrate, which does not come through here); a client turns
369            // memory on or off with `/memory`, not by create config.
370            config.memory_mode = None;
371            if config.cwd.is_none() {
372                config.cwd = Some(hello.cwd.clone());
373            }
374            if config.env.is_none() {
375                config.env = hello.env.clone();
376            }
377            if config.env_stripped.is_empty() {
378                config.env_stripped = hello.env_stripped.clone();
379            }
380            if let Some(cwd) = &config.cwd {
381                if !cwd.is_absolute() || !cwd.is_dir() {
382                    let _ = tx
383                        .send(DaemonFrame::Error {
384                            session_id: None,
385                            message: format!("cwd must be an absolute existing directory: {}", cwd.display()),
386                        })
387                        .await;
388                    let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
389                    let _ = writer.await;
390                    return;
391                }
392            }
393            // `--continue X` while X is live here: the host hands back the
394            // running actor (attach-if-live). Tell the client — it asked to
395            // create and got the existing session instead. Detected by id
396            // presence before/after so a client-less Parked target counts too.
397            let before: Vec<SessionId> = if config.continue_session.is_some() {
398                state.live_sessions().into_iter().map(|h| h.id).collect()
399            } else {
400                Vec::new()
401            };
402            match state.create(config).await {
403                Ok(h) if before.contains(&h.id) => {
404                    tracing::info!(session = %h.id, "daemon: continue target is live — attaching, not creating");
405                    attached_to_live = true;
406                    (h, mode)
407                }
408                Ok(h) => (h, mode),
409                Err(e) => {
410                    let _ = tx.send(DaemonFrame::Error { session_id: None, message: format!("create session: {e}") }).await;
411                    let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
412                    let _ = writer.await;
413                    return;
414                }
415            }
416        }
417    };
418
419    let mut rx = handle.subscribe();
420    if let Err(e) = handle.send(SessionCommand::Attach { client: hello.client.clone(), mode }).await {
421        let _ = tx.send(DaemonFrame::Error { session_id: Some(handle.id.clone()), message: format!("attach: {e}") }).await;
422        let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
423        let _ = writer.await;
424        return;
425    }
426    let attach_budget = if handle.lifecycle() == SessionLifecycle::Parked { ATTACH_TIMEOUT_PARKED } else { ATTACH_TIMEOUT };
427    let attached = tokio::time::timeout(attach_budget, async {
428        loop {
429            match rx.recv().await {
430                Ok(env) => match env.event {
431                    SessionEventWire::Attached { client, snapshot } => return Some(Ok((client, snapshot))),
432                    SessionEventWire::AttachRefused { message } => return Some(Err(message)),
433                    _ => {}
434                },
435                Err(broadcast::error::RecvError::Lagged(_)) => continue,
436                Err(broadcast::error::RecvError::Closed) => return None,
437            }
438        }
439    })
440    .await;
441    let (client, snapshot) = match attached {
442        Ok(Some(Ok(x))) => x,
443        Ok(Some(Err(message))) => {
444            let _ = tx.send(DaemonFrame::Error { session_id: Some(handle.id.clone()), message }).await;
445            let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
446            let _ = writer.await;
447            return;
448        }
449        _ => {
450            let _ = tx.send(DaemonFrame::Error { session_id: Some(handle.id.clone()), message: "attach timed out".into() }).await;
451            let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
452            let _ = writer.await;
453            return;
454        }
455    };
456    let legacy = legacy_shared(hello.client.capabilities, snapshot.client_input);
457    let snapshot = replay_for(hello.client.capabilities, snapshot);
458    let snapshot = if legacy { as_owner(client, snapshot) } else { snapshot };
459    if tx.send(DaemonFrame::Attached(AttachedWire::new(client, snapshot))).await.is_err() {
460        let _ = handle.send(SessionCommand::Detach { client }).await;
461        return;
462    }
463    tracing::debug!(session = %handle.id, client = client.0, "daemon: client attached");
464    if attached_to_live {
465        let env = Envelope {
466            session_id: handle.id.clone(),
467            seq: u64::MAX,
468            ts: chrono::Utc::now(),
469            event: SessionEventWire::SystemNotice(format!(
470                "session {} is already running in this daemon — attached to it instead of creating a second copy",
471                handle.id
472            )),
473        };
474        let _ = tx.send(DaemonFrame::Event(env.into())).await;
475    }
476
477    // ── pump ──
478    let sid = handle.id.clone();
479    let fwd_tx = tx.clone();
480    let fwd_handle = handle.clone();
481    let fwd_announce = state.announce();
482    // Phase 4 §2.3: Digest clients never receive the per-round full history;
483    // the `Conversation` digest the actor emits right after is what they key on.
484    let history = hello.client.history;
485    let caps = hello.client.capabilities;
486    let skip = move |env: &Envelope| -> bool {
487        !forwarded(history, caps, env) || (legacy && matches!(env.event, SessionEventWire::InputOwnerChanged { .. }))
488    };
489    // Not tied to `shutdown`: on daemon stop the client must still see `Ended`.
490    let mut forward = tokio::spawn(async move {
491        loop {
492            let next = tokio::select! {
493                biased;
494                r = rx.recv() => r,
495                // C3: reload announced — the checkpoint's events (abort
496                // notice, Conversation with the kept partial history) are already in
497                // the broadcast; drain what is there, then stop.
498                _ = fwd_announce.cancelled() => {
499                    while let Ok(env) = rx.try_recv() {
500                        if skip(&env) {
501                            continue;
502                        }
503                        if fwd_tx.send(DaemonFrame::Event(env.into())).await.is_err() {
504                            break;
505                        }
506                    }
507                    break;
508                }
509            };
510            match next {
511                Ok(env) => {
512                    // Attached is addressed to one client; ours already went out as a frame.
513                    if skip(&env) {
514                        continue;
515                    }
516                    let ended = matches!(env.event, SessionEventWire::Ended { .. });
517                    if fwd_tx.send(DaemonFrame::Event(env.into())).await.is_err() {
518                        break;
519                    }
520                    if ended {
521                        break;
522                    }
523                }
524                Err(broadcast::error::RecvError::Lagged(n)) => {
525                    tracing::warn!(session = %fwd_handle.id, dropped = n, "daemon: client lagged");
526                    let env = Envelope {
527                        session_id: fwd_handle.id.clone(),
528                        seq: u64::MAX,
529                        ts: chrono::Utc::now(),
530                        event: SessionEventWire::SystemNotice(format!("event stream lagged; {n} dropped")),
531                    };
532                    if fwd_tx.send(DaemonFrame::Event(env.into())).await.is_err() {
533                        break;
534                    }
535                }
536                Err(broadcast::error::RecvError::Closed) => break,
537            }
538        }
539    });
540
541    let mut ended = false;
542    let mut stopping = false;
543    let mut reloading = false;
544    let announce = state.announce();
545    loop {
546        let frame = tokio::select! {
547            _ = shutdown.cancelled() => { stopping = true; break; }
548            _ = announce.cancelled() => { reloading = true; break; }
549            _ = handle.closed() => { ended = true; break; }
550            r = read_frame(&mut reader, &mut line) => r,
551        };
552        match frame {
553            Ok(Read::Frame(ClientFrame::Cmd { session_id, cmd })) => {
554                if session_id != sid {
555                    let _ = tx.send(DaemonFrame::Error { session_id: Some(session_id), message: "not attached to that session".into() }).await;
556                    continue;
557                }
558                if let Err(why) = client_may_send(&cmd, client) {
559                    let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: format!("refused: {why}") }).await;
560                    continue;
561                }
562                if state.reloading.load(Ordering::SeqCst) && refused_while_draining(&cmd) {
563                    let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: crate::session::RELOADING_REFUSAL.into() }).await;
564                    continue;
565                }
566                match handle.send_from(client, cmd).await {
567                    Ok(()) => {}
568                    Err(TransportError::Backpressure) => {
569                        let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "backpressure: command queue full".into() }).await;
570                    }
571                    Err(TransportError::Closed) => {
572                        ended = true;
573                        break;
574                    }
575                    Err(e) => {
576                        let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: e.to_string() }).await;
577                    }
578                }
579            }
580            Ok(Read::Frame(ClientFrame::Ping)) => {
581                // `daemon status` and `daemon reload` read daemon.json after a Ping.
582                state.heal_daemon_json();
583                let _ = tx
584                    .send(DaemonFrame::Pong { pid: std::process::id(), uptime_s: state.uptime_s(), sessions: state.live_sessions().len() })
585                    .await;
586            }
587            Ok(Read::Frame(ClientFrame::Sessions)) => {
588                let _ = tx.send(DaemonFrame::SessionList { sessions: state.session_metas() }).await;
589            }
590            Ok(Read::Frame(ClientFrame::Shutdown { force })) => {
591                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
592                state.request_shutdown(force);
593                break;
594            }
595            Ok(Read::Frame(ClientFrame::Attach(_))) | Ok(Read::Frame(ClientFrame::Hello(_))) => {
596                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "already attached".into() }).await;
597            }
598            // An attached client closes its session with `/close` and pins
599            // it with `/pin`; `synaps auth state` is a control connection.
600            Ok(Read::Frame(ClientFrame::Reload { .. }))
601            | Ok(Read::Frame(ClientFrame::Purge))
602            | Ok(Read::Frame(ClientFrame::Close { .. }))
603            | Ok(Read::Frame(ClientFrame::Pin { .. }))
604            | Ok(Read::Frame(ClientFrame::ReloadExtension { .. }))
605            | Ok(Read::Frame(ClientFrame::AuthState(_))) => {
606                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "control frames are not accepted on an attached connection".into() }).await;
607            }
608            Ok(Read::Frame(ClientFrame::Bye)) | Ok(Read::Eof) | Err(_) => break,
609            Ok(Read::Oversize) => {
610                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: frame_limit_msg() }).await;
611                break;
612            }
613            Ok(Read::Bad(e)) => {
614                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: format!("malformed frame: {e}") }).await;
615            }
616        }
617    }
618
619    // C3: reload — the session is checkpointed and will be rehydrated by
620    // the next image; tell the client to reconnect and stop forwarding.
621    if reloading {
622        // The forwarder drains and exits on the announce; bound it anyway.
623        if tokio::time::timeout(Duration::from_secs(1), &mut forward).await.is_err() {
624            forward.abort();
625            let _ = forward.await;
626        }
627        let (generation, retry_after_ms) = match reload_bye(&state) {
628            ByeReason::Reloading { generation, retry_after_ms } => (generation, retry_after_ms),
629            _ => unreachable!(),
630        };
631        let env = crate::session::Envelope {
632            session_id: sid.clone(),
633            seq: u64::MAX,
634            ts: chrono::Utc::now(),
635            event: SessionEventWire::Reloading { generation, retry_after_ms },
636        };
637        let _ = tx.send(DaemonFrame::Event(env.into())).await;
638        let _ = tx.send(DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation, retry_after_ms }) }).await;
639        let _ = tokio::time::timeout(Duration::from_secs(1), writer).await;
640        return;
641    }
642
643    // Socket gone / Bye = Detach. The turn keeps running (§8).
644    if stopping {
645        // Daemon stop: let the forwarder deliver Ended inside the lifecycle budget.
646        match tokio::time::timeout(super::lifecycle::SESSION_END_BUDGET + Duration::from_secs(1), &mut forward).await {
647            Ok(_) => {}
648            Err(_) => {
649                forward.abort();
650                let _ = forward.await;
651            }
652        }
653    } else if ended {
654        // The actor broadcast `Ended` before it dropped its queue, so it is
655        // in the forwarder's receiver: let it go out before the `Bye`, or the
656        // client sees EOF and never learns the session ended (`/close` from
657        // another client, `daemon close`, eviction).
658        if tokio::time::timeout(Duration::from_secs(1), &mut forward).await.is_err() {
659            forward.abort();
660            let _ = forward.await;
661        }
662    } else {
663        forward.abort();
664        let _ = forward.await;
665    }
666    if !ended && handle.is_alive() {
667        let _ = handle.send(SessionCommand::Detach { client }).await;
668    }
669    if !handle.is_alive() {
670        state.remove(&sid);
671    }
672    let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
673    drop(tx);
674    let _ = tokio::time::timeout(Duration::from_millis(500), writer).await;
675    tracing::debug!(session = %sid, client = client.0, "daemon: client detached");
676}
677
678/// Whether the pump forwards `env` to this connection's client. Never the
679/// actor's `Attached` (the conn sends its own). Digest clients never get the
680/// per-round full history; the `Conversation` digest the actor emits right
681/// after is what they key on (phase 4 §2.3). And never an event the client
682/// cannot decode ([`client_decodes`]: an older client fails the whole frame).
683fn forwarded(history: HistoryMode, caps: ClientCapabilities, env: &Envelope) -> bool {
684    !(matches!(env.event, SessionEventWire::Attached { .. })
685        || (history == HistoryMode::Digest
686            && matches!(
687                env.event,
688                SessionEventWire::Stream(StreamEvent::Session(SessionEvent::MessageHistory(_)))
689            ))
690        || !client_decodes(caps, &env.event))
691}
692
693/// The attach snapshot as this client may receive it: its turn replay
694/// without events it cannot decode — one would sink the whole `Attached`
695/// frame, and with it every reconnect.
696fn replay_for(caps: ClientCapabilities, mut snapshot: AttachSnapshot) -> AttachSnapshot {
697    snapshot.replay.retain(|env| client_decodes(caps, &env.event));
698    snapshot
699}
700
701/// [`legacy_shared`]: the snapshot as an older client on a shared session
702/// gets it: it owns input unless it observes, and its replay has no owner
703/// changes (it never gets live ones either).
704fn as_owner(client: ClientId, mut snapshot: AttachSnapshot) -> AttachSnapshot {
705    let observes = snapshot.roster.iter().any(|c| c.id == client && c.mode == AttachMode::Observe);
706    if !observes {
707        snapshot.input_owner = Some(client);
708    }
709    snapshot.replay.retain(|env| !matches!(env.event, SessionEventWire::InputOwnerChanged { .. }));
710    snapshot
711}
712
713fn reload_bye(state: &DaemonState) -> ByeReason {
714    ByeReason::Reloading {
715        generation: state.reload_generation.load(Ordering::SeqCst),
716        retry_after_ms: super::reload::RETRY_AFTER_MS,
717    }
718}
719
720/// `AuthState`: the in-memory account state of the broker on the host's
721/// token cache, which every session of this daemon uses. Same trust as the
722/// other control frames (the socket is the user's own). No network request,
723/// no refresh; a clear never writes the credential store.
724fn auth_state_reply(state: &DaemonState, query: &agent_core::auth::AuthStateQuery) -> DaemonFrame {
725    match agent_core::auth::local_auth_state(&state.host.parts().token_cache, query, true) {
726        Ok(report) => DaemonFrame::AuthState { report },
727        Err(e) => DaemonFrame::Error { session_id: None, message: format!("auth state: {e}") },
728    }
729}
730
731#[cfg(test)]
732mod tests {
733    use super::*;
734    use crate::session::CheckpointReason;
735
736    /// A reload checkpoint stops every worker and fences the session until
737    /// an exec that only the daemon's own reload performs: no client may
738    /// send one. A host-request checkpoint stays a client command.
739    #[test]
740    fn a_reload_checkpoint_is_host_only() {
741        let own = ClientId(1);
742        let reload = SessionCommand::Checkpoint { reason: CheckpointReason::Reload };
743        assert!(client_may_send(&reload, own).is_err());
744        let host = SessionCommand::Checkpoint { reason: CheckpointReason::HostRequest };
745        assert!(client_may_send(&host, own).is_ok());
746    }
747
748    /// `/takeover` comes from a client by definition (the actor checks the
749    /// attach mode), so the conn lets it through.
750    #[test]
751    fn take_input_is_a_client_command() {
752        assert!(client_may_send(&SessionCommand::TakeInput, ClientId(2)).is_ok());
753    }
754
755    fn stream_env(ev: StreamEvent) -> Envelope {
756        Envelope {
757            session_id: SessionId::from("s"),
758            seq: 1,
759            ts: chrono::Utc::now(),
760            event: SessionEventWire::Stream(ev),
761        }
762    }
763
764    /// An older client (no `tool_queue` capability) is never sent the queue
765    /// signals — live or in the attach replay — and still gets every other
766    /// event; a current client gets them all.
767    #[test]
768    fn queue_signals_are_only_forwarded_to_clients_that_opted_in() {
769        use crate::LlmEvent;
770        let queued = stream_env(StreamEvent::Llm(LlmEvent::ToolQueued { tool_id: "t2".into() }));
771        let started = stream_env(StreamEvent::Llm(LlmEvent::ToolStarted { tool_id: "t2".into() }));
772        let result = stream_env(StreamEvent::Llm(LlmEvent::ToolResult {
773            tool_id: "t1".into(),
774            result: "ok".into(),
775        }));
776        let old = ClientCapabilities::default();
777        let new = ClientCapabilities::current();
778        for history in [HistoryMode::Full, HistoryMode::Digest] {
779            assert!(!forwarded(history, old, &queued));
780            assert!(!forwarded(history, old, &started));
781            assert!(forwarded(history, old, &result));
782            assert!(forwarded(history, new, &queued));
783            assert!(forwarded(history, new, &started));
784            assert!(forwarded(history, new, &result));
785        }
786
787        let session = crate::session::handle::echo::meta_for(&SessionId::from("s"));
788        let snapshot = AttachSnapshot {
789            meta: session,
790            view: crate::session::handle::echo::view_for(),
791            conversation: Default::default(),
792            streaming: true,
793            replay: vec![queued.clone(), result.clone(), started.clone()],
794            pending_prompts: vec![],
795            clients: vec![],
796            roster: Vec::new(),
797            client_input: ClientInput::wire_default(),
798            input_owner: None,
799            display_tail: None,
800        };
801        let kinds = |s: &AttachSnapshot| -> Vec<String> {
802            s.replay
803                .iter()
804                .map(|e| match &e.event {
805                    SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::ToolQueued { .. })) => "queued",
806                    SessionEventWire::Stream(StreamEvent::Llm(LlmEvent::ToolStarted { .. })) => "started",
807                    _ => "other",
808                })
809                .map(str::to_string)
810                .collect()
811        };
812        assert_eq!(kinds(&replay_for(old, snapshot.clone())), vec!["other"]);
813        assert_eq!(kinds(&replay_for(new, snapshot)), vec!["queued", "other", "started"]);
814    }
815
816    /// The drain gate refuses work that would start after the record, and
817    /// leaves what the turn being drained still needs.
818    #[test]
819    fn the_drain_gate_keeps_steering_the_running_turn() {
820        let submit = SessionCommand::Submit { text: "x".into(), attachments: vec![] };
821        assert!(refused_while_draining(&submit));
822        assert!(refused_while_draining(&SessionCommand::NewSession));
823        assert!(refused_while_draining(&SessionCommand::Resume { id: 1, query: "q".into() }));
824        assert!(!refused_while_draining(&SessionCommand::Steer { text: "x".into() }));
825        assert!(!refused_while_draining(&SessionCommand::Cancel));
826    }
827}
agent_engine/daemon/conn.rs

agent_engine/daemon/
conn.rs
