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 (`End`) is refused with an `Error` frame. Ending a session
33/// is `daemon stop`'s job today.
34fn client_may_send(cmd: &SessionCommand, own: ClientId) -> Result<(), &'static str> {
35    use SessionCommand as C;
36    match cmd {
37        C::Submit { .. }
38        | C::Steer { .. }
39        | C::Cancel
40        | C::Answer { .. }
41        | C::Set { .. }
42        | C::Compact { .. }
43        | C::NewSession
44        | C::Save
45        | C::Query { .. }
46        | C::EngineCommand { .. }
47        | C::SubmitPrepared { .. }
48        | C::PluginCommand { .. }
49        | C::Resume { .. }
50        | C::Checkpoint { .. }
51        | C::KeepWarm { .. } => Ok(()),
52        C::Detach { client } if *client == own => Ok(()),
53        C::Detach { .. } => Err("detach: not your client id"),
54        C::Attach { .. } => Err("attach: already attached (one session per connection)"),
55        // C4: an owner may end its session (`/end`); the actor enforces
56        // ownership (B1 `is_input_command`), the conn only screens reasons.
57        C::End { reason: EndReason::ClientQuit } => Ok(()),
58        C::End { .. } => Err("end: only ClientQuit may come from a client (host reasons are the daemon's)"),
59        C::Resync { .. } => Err("resync: not a client command"),
60        C::Park => Err("park: not a client command"),
61        C::HostEvent(_) => Err("host_event: not a client command"),
62        C::DriverStart { .. } => Ok(()),
63    }
64}
65
66enum Read {
67    Frame(ClientFrame),
68    Eof,
69    Oversize,
70    Bad(String),
71}
72
73async fn read_frame(reader: &mut Reader, line: &mut String) -> std::io::Result<Read> {
74    loop {
75        line.clear();
76        let n = tokio::io::AsyncReadExt::take(&mut *reader, MAX_FRAME_BYTES as u64 + 1)
77            .read_line(line)
78            .await?;
79        if n == 0 {
80            return Ok(Read::Eof);
81        }
82        if n > MAX_FRAME_BYTES {
83            return Ok(Read::Oversize);
84        }
85        let t = line.trim_end();
86        if t.is_empty() {
87            continue;
88        }
89        return Ok(match decode_line::<ClientFrame>(t) {
90            Ok(f) => Read::Frame(f),
91            Err(e) => Read::Bad(e),
92        });
93    }
94}
95
96/// Writer task: owns the write half; bounded queue = per-client backpressure.
97fn spawn_writer(mut w: OwnedWriteHalf) -> (mpsc::Sender<DaemonFrame>, tokio::task::JoinHandle<()>) {
98    let (tx, mut rx) = mpsc::channel::<DaemonFrame>(CMD_CHAN_CAP);
99    let task = tokio::spawn(async move {
100        while let Some(frame) = rx.recv().await {
101            let bye = matches!(frame, DaemonFrame::Bye { .. } | DaemonFrame::Refused { .. });
102            match encode_line(&frame) {
103                Ok(line) => {
104                    if w.write_all(line.as_bytes()).await.is_err() {
105                        break;
106                    }
107                }
108                Err(e) => {
109                    // Symmetric cap: an outbound frame we cannot legally send
110                    // becomes an Error frame and the connection closes.
111                    tracing::warn!("daemon: encode failed: {e}");
112                    let err = DaemonFrame::Error { session_id: None, message: format!("daemon could not encode a frame: {e}") };
113                    if let Ok(line) = encode_line(&err) {
114                        let _ = w.write_all(line.as_bytes()).await;
115                    }
116                    let _ = w.shutdown().await;
117                    break;
118                }
119            }
120            if bye {
121                let _ = w.shutdown().await;
122                break;
123            }
124        }
125    });
126    (tx, task)
127}
128
129pub async fn serve(state: Arc<DaemonState>, stream: UnixStream, shutdown: CancellationToken) {
130    let (r, w) = stream.into_split();
131    let mut reader = BufReader::new(r);
132    let (tx, writer) = spawn_writer(w);
133    let mut line = String::new();
134
135    // ── handshake: first frame MUST be Hello ──
136    let hello = match tokio::time::timeout(HELLO_TIMEOUT, read_frame(&mut reader, &mut line)).await {
137        Ok(Ok(Read::Frame(ClientFrame::Hello(h)))) => h,
138        Ok(Ok(Read::Frame(_))) | Ok(Ok(Read::Bad(_))) => {
139            let _ = tx
140                .send(DaemonFrame::Refused { reason: RefuseReason::Protocol, message: "first frame must be hello".into() })
141                .await;
142            let _ = writer.await;
143            return;
144        }
145        Ok(Ok(Read::Oversize)) => {
146            let _ = tx.send(DaemonFrame::Error { session_id: None, message: frame_limit_msg() }).await;
147            drop(tx);
148            let _ = writer.await;
149            return;
150        }
151        _ => {
152            drop(tx);
153            let _ = writer.await;
154            return;
155        }
156    };
157    if hello.protocol_version < PROTOCOL_MIN || hello.protocol_version > PROTOCOL_MAX {
158        tracing::warn!(client = hello.protocol_version, daemon = PROTOCOL_VERSION, "daemon: refusing protocol version");
159        let _ = tx
160            .send(DaemonFrame::Refused {
161                reason: RefuseReason::Version { daemon_version: PROTOCOL_VERSION, min: PROTOCOL_MIN, max: PROTOCOL_MAX },
162                message: format!(
163                    "protocol version {} not supported (daemon speaks {}; daemon binary {})",
164                    hello.protocol_version,
165                    PROTOCOL_VERSION,
166                    binary_version()
167                ),
168            })
169            .await;
170        let _ = writer.await;
171        return;
172    }
173    let daemon_version = binary_version();
174    if hello.client_version != daemon_version {
175        tracing::info!(client = %hello.client_version, daemon = %daemon_version, "daemon: client binary version differs (same protocol; allowed)");
176    }
177    let welcome = Welcome {
178        protocol_version: PROTOCOL_VERSION,
179        daemon_version,
180        pid: std::process::id(),
181        profile: state.profile.clone(),
182        sessions: state.session_metas(),
183        progressive_tool_disclosure: state.host.parts().progressive_tool_disclosure,
184        generation: state.generation,
185    };
186    if tx.send(DaemonFrame::Welcome(welcome)).await.is_err() {
187        return;
188    }
189    tracing::debug!(kind = ?hello.client.kind, "daemon: client connected");
190
191    // ── control loop (no session allocated) ──
192    let attach = loop {
193        let announce = state.announce();
194        let frame = tokio::select! {
195            _ = shutdown.cancelled() => { let _ = tx.send(DaemonFrame::Bye { reason: None }).await; let _ = writer.await; return; }
196            _ = announce.cancelled() => {
197                let _ = tx.send(DaemonFrame::Bye { reason: Some(reload_bye(&state)) }).await;
198                let _ = writer.await;
199                return;
200            }
201            r = read_frame(&mut reader, &mut line) => r,
202        };
203        match frame {
204            Ok(Read::Frame(ClientFrame::Ping)) => {
205                let _ = tx
206                    .send(DaemonFrame::Pong { pid: std::process::id(), uptime_s: state.uptime_s(), sessions: state.live_sessions().len() })
207                    .await;
208            }
209            Ok(Read::Frame(ClientFrame::Sessions)) => {
210                let _ = tx.send(DaemonFrame::SessionList { sessions: state.session_metas() }).await;
211            }
212            Ok(Read::Frame(ClientFrame::Shutdown { force })) => {
213                tracing::info!(force, "daemon: shutdown requested over the socket");
214                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
215                let _ = writer.await;
216                state.request_shutdown(force);
217                return;
218            }
219            Ok(Read::Frame(ClientFrame::Attach(a))) => {
220                if matches!(a, Attach::Create { .. }) && state.reloading.load(Ordering::SeqCst) {
221                    let _ = tx
222                        .send(DaemonFrame::Refused { reason: RefuseReason::Busy, message: "daemon reloading; retry in a moment".into() })
223                        .await;
224                    continue;
225                }
226                break a;
227            }
228            Ok(Read::Frame(ClientFrame::Cmd { session_id, .. })) => {
229                let _ = tx.send(DaemonFrame::Error { session_id: Some(session_id), message: "not attached".into() }).await;
230            }
231            // C3: the whole §2.8 sequence runs on this connection's task;
232            // on success the process is replaced and this never returns.
233            Ok(Read::Frame(ref f @ ClientFrame::Reload { .. })) => {
234                let req = super::reload::ReloadRequest::from_frame(f).expect("reload frame");
235                match super::reload::prepare(&state, &state.paths, req).await {
236                    Err(super::reload::ReloadError::Refused(why)) => {
237                        let _ = tx
238                            .send(DaemonFrame::Refused { reason: RefuseReason::ReloadRefused { why: why.clone() }, message: why })
239                            .await;
240                    }
241                    Err(super::reload::ReloadError::ExecFailed(e)) => {
242                        let _ = tx.send(DaemonFrame::Error { session_id: None, message: format!("reload: {e}") }).await;
243                    }
244                    Ok(prepared) => {
245                        // Tell the requester, flush, close — then exec.
246                        let _ = tx.send(DaemonFrame::Bye { reason: Some(reload_bye(&state)) }).await;
247                        drop(tx);
248                        let _ = tokio::time::timeout(Duration::from_secs(1), writer).await;
249                        let e = super::reload::exec(&state, &state.paths, prepared).await;
250                        tracing::error!(error = %e, "daemon: reload did not exec");
251                        return;
252                    }
253                }
254            }
255            // C2: jemalloc purge in the daemon (bench hygiene); reply Pong.
256            Ok(Read::Frame(ClientFrame::Purge)) => {
257                agent_core::core::memstat::purge_arenas();
258                let _ = tx
259                    .send(DaemonFrame::Pong { pid: std::process::id(), uptime_s: state.uptime_s(), sessions: state.live_sessions().len() })
260                    .await;
261            }
262            Ok(Read::Frame(ClientFrame::Hello(_))) => {
263                let _ = tx.send(DaemonFrame::Error { session_id: None, message: "duplicate hello".into() }).await;
264            }
265            Ok(Read::Frame(ClientFrame::Bye)) | Ok(Read::Eof) | Err(_) => {
266                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
267                let _ = writer.await;
268                return;
269            }
270            Ok(Read::Oversize) => {
271                let _ = tx.send(DaemonFrame::Error { session_id: None, message: frame_limit_msg() }).await;
272                drop(tx);
273                let _ = writer.await;
274                return;
275            }
276            Ok(Read::Bad(e)) => {
277                let _ = tx.send(DaemonFrame::Error { session_id: None, message: format!("malformed frame: {e}") }).await;
278            }
279        }
280    };
281
282    // ── attach ──
283    let mut attached_to_live = false;
284    let (handle, mode) = match attach {
285        Attach::Existing { session_id, mode } => match state.attach(&session_id) {
286            Some(h) if h.meta().locked_by.is_some() => (state.retry_locked_placeholder(h).await, mode),
287            Some(h) => (h, mode),
288            None => {
289                let _ = tx
290                    .send(DaemonFrame::Error { session_id: Some(session_id), message: "unknown session".into() })
291                    .await;
292                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
293                let _ = writer.await;
294                return;
295            }
296        },
297        Attach::Create { mut config, mode } => {
298            if config.cwd.is_none() {
299                config.cwd = Some(hello.cwd.clone());
300            }
301            if config.env.is_none() {
302                config.env = hello.env.clone();
303            }
304            if config.env_stripped.is_empty() {
305                config.env_stripped = hello.env_stripped.clone();
306            }
307            if let Some(cwd) = &config.cwd {
308                if !cwd.is_absolute() || !cwd.is_dir() {
309                    let _ = tx
310                        .send(DaemonFrame::Error {
311                            session_id: None,
312                            message: format!("cwd must be an absolute existing directory: {}", cwd.display()),
313                        })
314                        .await;
315                    let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
316                    let _ = writer.await;
317                    return;
318                }
319            }
320            // `--continue X` while X is live here: the host hands back the
321            // running actor (attach-if-live). Tell the client — it asked to
322            // create and got the existing session instead. Detected by id
323            // presence before/after so a client-less Parked target counts too.
324            let before: Vec<SessionId> = if config.continue_session.is_some() {
325                state.live_sessions().into_iter().map(|h| h.id).collect()
326            } else {
327                Vec::new()
328            };
329            match state.create(config).await {
330                Ok(h) if before.contains(&h.id) => {
331                    tracing::info!(session = %h.id, "daemon: continue target is live — attaching, not creating");
332                    attached_to_live = true;
333                    (h, mode)
334                }
335                Ok(h) => (h, mode),
336                Err(e) => {
337                    let _ = tx.send(DaemonFrame::Error { session_id: None, message: format!("create session: {e}") }).await;
338                    let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
339                    let _ = writer.await;
340                    return;
341                }
342            }
343        }
344    };
345
346    let mut rx = handle.subscribe();
347    if let Err(e) = handle.send(SessionCommand::Attach { client: hello.client.clone(), mode }).await {
348        let _ = tx.send(DaemonFrame::Error { session_id: Some(handle.id.clone()), message: format!("attach: {e}") }).await;
349        let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
350        let _ = writer.await;
351        return;
352    }
353    let attach_budget = if handle.lifecycle() == SessionLifecycle::Parked { ATTACH_TIMEOUT_PARKED } else { ATTACH_TIMEOUT };
354    let attached = tokio::time::timeout(attach_budget, async {
355        loop {
356            match rx.recv().await {
357                Ok(env) => match env.event {
358                    SessionEventWire::Attached { client, snapshot } => return Some(Ok((client, snapshot))),
359                    SessionEventWire::AttachRefused { message } => return Some(Err(message)),
360                    _ => {}
361                },
362                Err(broadcast::error::RecvError::Lagged(_)) => continue,
363                Err(broadcast::error::RecvError::Closed) => return None,
364            }
365        }
366    })
367    .await;
368    let (client, snapshot) = match attached {
369        Ok(Some(Ok(x))) => x,
370        Ok(Some(Err(message))) => {
371            let _ = tx.send(DaemonFrame::Error { session_id: Some(handle.id.clone()), message }).await;
372            let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
373            let _ = writer.await;
374            return;
375        }
376        _ => {
377            let _ = tx.send(DaemonFrame::Error { session_id: Some(handle.id.clone()), message: "attach timed out".into() }).await;
378            let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
379            let _ = writer.await;
380            return;
381        }
382    };
383    if tx.send(DaemonFrame::Attached(AttachedWire::new(client, snapshot))).await.is_err() {
384        let _ = handle.send(SessionCommand::Detach { client }).await;
385        return;
386    }
387    tracing::debug!(session = %handle.id, client = client.0, "daemon: client attached");
388    if attached_to_live {
389        let env = Envelope {
390            session_id: handle.id.clone(),
391            seq: u64::MAX,
392            ts: chrono::Utc::now(),
393            event: SessionEventWire::SystemNotice(format!(
394                "session {} is already running in this daemon — attached to it instead of creating a second copy",
395                handle.id
396            )),
397        };
398        let _ = tx.send(DaemonFrame::Event(env.into())).await;
399    }
400
401    // ── pump ──
402    let sid = handle.id.clone();
403    let fwd_tx = tx.clone();
404    let fwd_handle = handle.clone();
405    let fwd_announce = state.announce();
406    // Phase 4 §2.3: Digest clients never receive the per-round full history;
407    // the `Conversation` digest the actor emits right after is what they key on.
408    let history = hello.client.history;
409    let skip = move |env: &Envelope| -> bool {
410        matches!(env.event, SessionEventWire::Attached { .. })
411            || (history == HistoryMode::Digest
412                && matches!(
413                    env.event,
414                    SessionEventWire::Stream(StreamEvent::Session(SessionEvent::MessageHistory(_)))
415                ))
416    };
417    // Not tied to `shutdown`: on daemon stop the client must still see `Ended`.
418    let mut forward = tokio::spawn(async move {
419        loop {
420            let next = tokio::select! {
421                biased;
422                r = rx.recv() => r,
423                // C3: reload announced — the checkpoint's events (abort
424                // notice, Conversation with the kept partial history) are already in
425                // the broadcast; drain what is there, then stop.
426                _ = fwd_announce.cancelled() => {
427                    while let Ok(env) = rx.try_recv() {
428                        if skip(&env) {
429                            continue;
430                        }
431                        if fwd_tx.send(DaemonFrame::Event(env.into())).await.is_err() {
432                            break;
433                        }
434                    }
435                    break;
436                }
437            };
438            match next {
439                Ok(env) => {
440                    // Attached is addressed to one client; ours already went out as a frame.
441                    if skip(&env) {
442                        continue;
443                    }
444                    let ended = matches!(env.event, SessionEventWire::Ended { .. });
445                    if fwd_tx.send(DaemonFrame::Event(env.into())).await.is_err() {
446                        break;
447                    }
448                    if ended {
449                        break;
450                    }
451                }
452                Err(broadcast::error::RecvError::Lagged(n)) => {
453                    tracing::warn!(session = %fwd_handle.id, dropped = n, "daemon: client lagged");
454                    let env = Envelope {
455                        session_id: fwd_handle.id.clone(),
456                        seq: u64::MAX,
457                        ts: chrono::Utc::now(),
458                        event: SessionEventWire::SystemNotice(format!("event stream lagged; {n} dropped")),
459                    };
460                    if fwd_tx.send(DaemonFrame::Event(env.into())).await.is_err() {
461                        break;
462                    }
463                }
464                Err(broadcast::error::RecvError::Closed) => break,
465            }
466        }
467    });
468
469    let mut ended = false;
470    let mut stopping = false;
471    let mut reloading = false;
472    let announce = state.announce();
473    loop {
474        let frame = tokio::select! {
475            _ = shutdown.cancelled() => { stopping = true; break; }
476            _ = announce.cancelled() => { reloading = true; break; }
477            _ = handle.closed() => { ended = true; break; }
478            r = read_frame(&mut reader, &mut line) => r,
479        };
480        match frame {
481            Ok(Read::Frame(ClientFrame::Cmd { session_id, cmd })) => {
482                if session_id != sid {
483                    let _ = tx.send(DaemonFrame::Error { session_id: Some(session_id), message: "not attached to that session".into() }).await;
484                    continue;
485                }
486                if let Err(why) = client_may_send(&cmd, client) {
487                    let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: format!("refused: {why}") }).await;
488                    continue;
489                }
490                if state.reloading.load(Ordering::SeqCst)
491                    && matches!(cmd, SessionCommand::Submit { .. } | SessionCommand::SubmitPrepared { .. } | SessionCommand::Compact { .. })
492                {
493                    let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "daemon reloading; retry in a moment".into() }).await;
494                    continue;
495                }
496                match handle.send_from(client, cmd).await {
497                    Ok(()) => {}
498                    Err(TransportError::Backpressure) => {
499                        let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "backpressure: command queue full".into() }).await;
500                    }
501                    Err(TransportError::Closed) => {
502                        ended = true;
503                        break;
504                    }
505                    Err(e) => {
506                        let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: e.to_string() }).await;
507                    }
508                }
509            }
510            Ok(Read::Frame(ClientFrame::Ping)) => {
511                let _ = tx
512                    .send(DaemonFrame::Pong { pid: std::process::id(), uptime_s: state.uptime_s(), sessions: state.live_sessions().len() })
513                    .await;
514            }
515            Ok(Read::Frame(ClientFrame::Sessions)) => {
516                let _ = tx.send(DaemonFrame::SessionList { sessions: state.session_metas() }).await;
517            }
518            Ok(Read::Frame(ClientFrame::Shutdown { force })) => {
519                let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
520                state.request_shutdown(force);
521                break;
522            }
523            Ok(Read::Frame(ClientFrame::Attach(_))) | Ok(Read::Frame(ClientFrame::Hello(_))) => {
524                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "already attached".into() }).await;
525            }
526            Ok(Read::Frame(ClientFrame::Reload { .. })) | Ok(Read::Frame(ClientFrame::Purge)) => {
527                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: "control frames are not accepted on an attached connection".into() }).await;
528            }
529            Ok(Read::Frame(ClientFrame::Bye)) | Ok(Read::Eof) | Err(_) => break,
530            Ok(Read::Oversize) => {
531                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: frame_limit_msg() }).await;
532                break;
533            }
534            Ok(Read::Bad(e)) => {
535                let _ = tx.send(DaemonFrame::Error { session_id: Some(sid.clone()), message: format!("malformed frame: {e}") }).await;
536            }
537        }
538    }
539
540    // C3: reload — the session is checkpointed and will be rehydrated by
541    // the next image; tell the client to reconnect and stop forwarding.
542    if reloading {
543        // The forwarder drains and exits on the announce; bound it anyway.
544        if tokio::time::timeout(Duration::from_secs(1), &mut forward).await.is_err() {
545            forward.abort();
546            let _ = forward.await;
547        }
548        let (generation, retry_after_ms) = match reload_bye(&state) {
549            ByeReason::Reloading { generation, retry_after_ms } => (generation, retry_after_ms),
550            _ => unreachable!(),
551        };
552        let env = crate::session::Envelope {
553            session_id: sid.clone(),
554            seq: u64::MAX,
555            ts: chrono::Utc::now(),
556            event: SessionEventWire::Reloading { generation, retry_after_ms },
557        };
558        let _ = tx.send(DaemonFrame::Event(env.into())).await;
559        let _ = tx.send(DaemonFrame::Bye { reason: Some(ByeReason::Reloading { generation, retry_after_ms }) }).await;
560        let _ = tokio::time::timeout(Duration::from_secs(1), writer).await;
561        return;
562    }
563
564    // Socket gone / Bye = Detach. The turn keeps running (§8).
565    if stopping {
566        // Daemon stop: let the forwarder deliver Ended inside the lifecycle budget.
567        match tokio::time::timeout(super::lifecycle::SESSION_END_BUDGET + Duration::from_secs(1), &mut forward).await {
568            Ok(_) => {}
569            Err(_) => {
570                forward.abort();
571                let _ = forward.await;
572            }
573        }
574    } else {
575        forward.abort();
576        let _ = forward.await;
577    }
578    if !ended && handle.is_alive() {
579        let _ = handle.send(SessionCommand::Detach { client }).await;
580    }
581    if !handle.is_alive() {
582        state.remove(&sid);
583    }
584    let _ = tx.send(DaemonFrame::Bye { reason: None }).await;
585    drop(tx);
586    let _ = tokio::time::timeout(Duration::from_millis(500), writer).await;
587    tracing::debug!(session = %sid, client = client.0, "daemon: client detached");
588}
589
590fn reload_bye(state: &DaemonState) -> ByeReason {
591    ByeReason::Reloading {
592        generation: state.reload_generation.load(Ordering::SeqCst),
593        retry_after_ms: super::reload::RETRY_AFTER_MS,
594    }
595}
agent_engine/daemon/conn.rs

agent_engine/daemon/
conn.rs
