agent_engine/daemon/
rename.rs

1//! `ClientFrame::Rename` (`synaps daemon rename <ID|NAME> <NEW_NAME> |
2//! --clear [--force]`): name or unname one session from a control
3//! connection, without attaching.
4//!
5//! Renaming is metadata, not input: it needs no input ownership, and the
6//! connection never attaches, so presence (clients, owner, the shared-inbox
7//! claim) does not change. The daemon sends the session the host-only
8//! engine command `rename` (JSON args: name + the compare-and-set `expect`),
9//! handled by `SessionActor::host_rename`: `/saveas`'s rules under a
10//! process-wide name lock, a durable save that keeps the file mtimes, and a
11//! `Conversation` to every attached client. A parked session is renamed on
12//! disk under its journal lock (`session_catalog::rename_past_session`)
13//! without being woken; it picks the name up from disk when it unparks.
14
15use std::sync::atomic::{AtomicU64, Ordering};
16use std::time::Duration;
17
18use tokio::sync::broadcast;
19
20use super::DaemonState;
21use crate::session::actor::RENAME_COMMAND;
22use crate::session::actor_cmds::RenameArgs;
23use crate::session::wire::{DaemonFrame, RefuseReason, RenameRefusal, RENAME_QUERY_ID_BASE};
24use crate::session::{SessionCommand, SessionEventWire, SessionId, TransportError};
25
26/// How long a full command queue is retried before giving up.
27pub const RENAME_SEND_RETRY: Duration = Duration::from_secs(2);
28/// How long the actor may take to answer: queue latency plus one durable
29/// save (a parked session's save folds its journal, which can be large).
30pub const RENAME_REPLY_BUDGET: Duration = Duration::from_secs(15);
31
32fn next_rename_query_id() -> u64 {
33    static NEXT: AtomicU64 = AtomicU64::new(0);
34    RENAME_QUERY_ID_BASE + NEXT.fetch_add(1, Ordering::Relaxed)
35}
36
37fn failed(session_id: SessionId, reason: RenameRefusal, message: String) -> DaemonFrame {
38    DaemonFrame::RenameFailed { session_id, reason, message, taken_by: None, current_name: None }
39}
40
41/// Serve `Rename{session_id, name, expect}`. `None` or blank clears the
42/// name. Always answers with one frame: `Renamed`, `RenameFailed`, or
43/// `Refused{UnknownSession}`.
44pub async fn rename_session(
45    state: &DaemonState,
46    id: &SessionId,
47    name: Option<&str>,
48    expect: Option<Option<String>>,
49) -> DaemonFrame {
50    if state.reloading.load(Ordering::SeqCst) {
51        return failed(id.clone(), RenameRefusal::Busy, "daemon reloading; retry in a moment".into());
52    }
53    let Some(handle) = state.attach(id) else {
54        return DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: format!("no live session {id}") };
55    };
56    let sid = handle.id.clone();
57    if let Some(holder) = handle.meta().locked_by.clone() {
58        return failed(
59            sid.clone(),
60            RenameRefusal::Locked,
61            format!("session {sid} is live in another process ({holder}), not in this daemon; rename it there with /saveas"),
62        );
63    }
64    let arg = serde_json::to_string(&RenameArgs { name: name.map(str::to_string), expect })
65        .expect("rename args serialize");
66
67    // Subscribe before sending: the reply can land before we would look.
68    let mut rx = handle.subscribe();
69    let query_id = next_rename_query_id();
70    let deadline = tokio::time::Instant::now() + RENAME_SEND_RETRY;
71    loop {
72        let cmd = SessionCommand::EngineCommand { id: query_id, name: RENAME_COMMAND.into(), arg: arg.clone() };
73        match handle.send(cmd).await {
74            Ok(()) => break,
75            Err(TransportError::Backpressure) if tokio::time::Instant::now() < deadline => {
76                tokio::time::sleep(Duration::from_millis(20)).await;
77            }
78            Err(TransportError::Backpressure) => {
79                return failed(sid, RenameRefusal::Busy, "the session's command queue stayed full; retry".into());
80            }
81            Err(_) => {
82                return DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: format!("session {sid} has ended") };
83            }
84        }
85    }
86
87    let reply = tokio::time::timeout(RENAME_REPLY_BUDGET, async {
88        loop {
89            match rx.recv().await {
90                Ok(env) => match env.event {
91                    SessionEventWire::QueryResult { id, value } if id == query_id => return Some(value),
92                    SessionEventWire::Ended { .. } => return None,
93                    _ => {}
94                },
95                Err(broadcast::error::RecvError::Lagged(_)) => continue,
96                Err(broadcast::error::RecvError::Closed) => return None,
97            }
98        }
99    })
100    .await;
101    match reply {
102        Ok(Some(value)) => reply_frame(sid, &value),
103        Ok(None) => DaemonFrame::Refused { reason: RefuseReason::UnknownSession, message: format!("session {sid} has ended") },
104        Err(_) => failed(
105            sid,
106            RenameRefusal::Unknown,
107            format!(
108                "no reply from the session in {} s; the rename may still apply. Re-list (`synaps daemon sessions`) before retrying",
109                RENAME_REPLY_BUDGET.as_secs()
110            ),
111        ),
112    }
113}
114
115/// The actor's `rename` reply → the wire answer.
116fn reply_frame(session_id: SessionId, value: &serde_json::Value) -> DaemonFrame {
117    let text = value.get("text").and_then(|t| t.as_str()).unwrap_or_default().to_string();
118    if value.get("kind").and_then(|k| k.as_str()) == Some("error") {
119        let code = value.get("code").and_then(|c| c.as_str()).unwrap_or("failed");
120        let taken_by = value.get("taken_by").and_then(|t| t.as_str()).map(str::to_string);
121        let current_name = value.get("current_name").and_then(|t| t.as_str()).map(str::to_string);
122        return DaemonFrame::RenameFailed {
123            session_id,
124            reason: RenameRefusal::from_code(code),
125            message: text,
126            taken_by,
127            current_name,
128        };
129    }
130    let name = value.get("name").and_then(|n| n.as_str()).map(str::to_string);
131    let in_memory_only = value.get("in_memory_only").and_then(|b| b.as_bool()).unwrap_or(false);
132    DaemonFrame::Renamed { session_id, name, in_memory_only }
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138
139    #[test]
140    fn actor_replies_map_to_typed_frames() {
141        let ok = serde_json::json!({ "kind": "output", "text": "session named 'axon'", "name": "axon" });
142        assert!(matches!(reply_frame("s".into(), &ok), DaemonFrame::Renamed { name: Some(ref n), in_memory_only: false, .. } if n == "axon"));
143        let cleared = serde_json::json!({ "kind": "output", "text": "x", "name": null, "in_memory_only": true });
144        assert!(matches!(reply_frame("s".into(), &cleared), DaemonFrame::Renamed { name: None, in_memory_only: true, .. }));
145        let named = serde_json::json!({ "kind": "error", "code": "named", "text": "x", "current_name": "kite" });
146        assert!(matches!(
147            reply_frame("s".into(), &named),
148            DaemonFrame::RenameFailed { reason: RenameRefusal::Named, current_name: Some(ref c), .. } if c == "kite"
149        ));
150        let taken = serde_json::json!({
151            "kind": "error", "code": "taken", "text": "name 'axon' already used by session t", "taken_by": "t",
152        });
153        assert!(matches!(
154            reply_frame("s".into(), &taken),
155            DaemonFrame::RenameFailed { reason: RenameRefusal::Taken, taken_by: Some(ref t), ref message, .. }
156                if t == "t" && message.contains("already used")
157        ));
158        let odd = serde_json::json!({ "kind": "error", "text": "x" });
159        assert!(matches!(reply_frame("s".into(), &odd), DaemonFrame::RenameFailed { reason: RenameRefusal::Failed, .. }));
160    }
161}
agent_engine/daemon/rename.rs

agent_engine/daemon/
rename.rs
