agent_engine/daemon/
pin.rs

1//! `ClientFrame::Pin` (`synaps daemon pin <ID|NAME> [--off]`): pin or unpin
2//! one session from a control connection, without attaching to it.
3//!
4//! The daemon sends the session `KeepWarm{on}` with host authority
5//! (`from: None`), so the actor's own handling runs unchanged: pinning a
6//! parked session unparks it, unpinning one leaves it parked. The answer
7//! comes from the handle's pin cell, which the actor writes when it applies
8//! the command; no reply event to match.
9
10use std::sync::Arc;
11use std::time::Duration;
12
13use super::DaemonState;
14use crate::session::wire::{DaemonFrame, RefuseReason};
15use crate::session::{SessionCommand, SessionId, SessionLifecycle, TransportError};
16
17/// How long the daemon waits for the actor to apply the pin (queue latency).
18pub const PIN_BUDGET: Duration = Duration::from_secs(2);
19/// Pinning a parked session rebuilds its runtime first (MCP servers,
20/// extensions), which can take longer than [`PIN_BUDGET`].
21pub const PIN_UNPARK_BUDGET: Duration = Duration::from_secs(10);
22/// How often the pin cell is polled while waiting.
23const PIN_POLL: Duration = Duration::from_millis(20);
24
25/// Serve `Pin{session_id, on}`. Always answers with one frame: `Pinned`,
26/// `Refused{Busy|UnknownSession}` or `Error`.
27pub async fn pin_session(state: &Arc<DaemonState>, id: &SessionId, on: bool) -> DaemonFrame {
28    // The actor's reload fence swallows a host command, so the pin would
29    // never land.
30    if state.reloading.load(std::sync::atomic::Ordering::SeqCst) {
31        return DaemonFrame::Refused { reason: RefuseReason::Busy, message: "daemon reloading; retry in a moment".into() };
32    }
33    // A pin that is not live here (waiting, failed, not restored yet, or a
34    // lock placeholder): `--off` clears its entry, on restores it now.
35    let live = state.attach(id);
36    if live.as_ref().is_none_or(|h| h.meta().locked_by.is_some() && !on) {
37        match super::pins::pin_offline(state, id.as_str(), on).await {
38            Some(Ok(())) => return DaemonFrame::Pinned { session_id: id.clone(), on },
39            Some(Err(message)) => return DaemonFrame::Error { session_id: Some(id.clone()), message },
40            None => {}
41        }
42    }
43    let Some(handle) = live else {
44        return DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: format!("no live session {id}") };
45    };
46    let sid = handle.id.clone();
47    if let Some(holder) = handle.meta().locked_by.clone() {
48        return DaemonFrame::Refused {
49            reason: RefuseReason::Busy,
50            // Usually an in-process TUI, which has no pin (`/pin` there only
51            // says it applies to daemon sessions): name the way into the
52            // daemon instead of sending the user back to it.
53            message: format!(
54                "session {sid} is live in another process ({holder}), not in this daemon, so there is nothing \
55                 to pin here. Quit it there, then load it into the daemon with `synaps --attach --continue {sid}` \
56                 and pin it with /pin or `synaps daemon pin {sid}`"
57            ),
58        };
59    }
60
61    let budget = if on && handle.lifecycle() == SessionLifecycle::Parked { PIN_UNPARK_BUDGET } else { PIN_BUDGET };
62    let deadline = tokio::time::Instant::now() + budget;
63    loop {
64        match handle.send(SessionCommand::KeepWarm { on }).await {
65            Ok(()) => break,
66            Err(TransportError::Backpressure) if tokio::time::Instant::now() < deadline => {
67                tokio::time::sleep(PIN_POLL).await;
68            }
69            Err(TransportError::Backpressure) => {
70                return DaemonFrame::Error {
71                    session_id: Some(sid),
72                    message: "pin: the session's command queue stayed full; retry".into(),
73                };
74            }
75            Err(_) => {
76                return DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: format!("session {sid} has ended") };
77            }
78        }
79    }
80    while handle.pinned() != on {
81        if !handle.is_alive() {
82            return DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: format!("session {sid} has ended") };
83        }
84        if tokio::time::Instant::now() >= deadline {
85            return DaemonFrame::Error {
86                session_id: Some(sid),
87                message: format!("pin: the session did not apply it within {} s", budget.as_secs()),
88            };
89        }
90        tokio::time::sleep(PIN_POLL).await;
91    }
92    DaemonFrame::Pinned { session_id: sid, on }
93}
agent_engine/daemon/pin.rs

agent_engine/daemon/
pin.rs
