Skip to main content

agent_engine/daemon/
reload.rs

1//! `synaps daemon reload` (PLAN-phase3 §2.8, C3): version gate → drain →
2//! `Checkpoint{Reload}` every session → reload-state → announce
3//! `Reloading`/`Bye{Reloading}` → `execv` self (same pid, adopted flock,
4//! CLOEXEC listener) → new image rehydrates sessions from reload-state
5//! BEFORE accepting.
6//!
7//! What reload cannot preserve (§2.8, stated once): in-flight turns
8//! (checkpointed = cancelled; partial history + restart marker kept), pending prompts (answered
9//! `None`), PTY/background shells (closed, announced), `turn_replay`,
10//! input ownership (the owner reclaims it on
11//! reconnect via `was_owner`). What it preserves — each session's
12//! `Checkpoint{Reload}` reply carries a `SessionReloadRecord`: its journal
13//! and id (the rehydrated session continues the same journal), its
14//! `SessionConfig` as created (cwd, system, prompt manifest, compaction
15//! policy, auto-compact, …), its keep-warm pin, its lifecycle (a Parked
16//! session comes back Parked), the non-persisted runtime knobs
17//! (`settings_replay`: context window, compaction model, retries,
18//! timeouts, tool-output cap, worker grants, `/system`), the CURRENT
19//! model/thinking (`/model` mid-session survives), and the session's exact
20//! tool activations (pins only, re-validated by the new runtime at its first
21//! turn start — `SessionToolSet::restore_carried`).
22//!
23//! Windows has no `execv`, and a byte-range lock cannot be inherited, so
24//! step 5 is a hand-off between two processes (see `daemon::win`): the old
25//! image starts the new one with the reload-state and a hand-off event; the
26//! new image queues its `daemon.lock` request behind the old one and sets
27//! the event; the old image exits — discarding what an exec would — and its
28//! release grants the queued request, so the liveness oracle never reads
29//! "free". The new image does this before booting anything, then proceeds
30//! exactly as after an exec. Same guarantees, one difference: the pid
31//! changes (clients learn it from the next `Welcome`).
32
33use std::path::{Path, PathBuf};
34use std::sync::atomic::Ordering;
35use std::sync::Arc;
36use std::time::Duration;
37
38use serde::{Deserialize, Serialize};
39
40use super::registry::{self, DaemonLock, DaemonPaths};
41use super::DaemonState;
42use crate::session::wire::{ClientFrame, CHECKPOINT_QUERY_ID, PROTOCOL_VERSION};
43use crate::session::{
44    CheckpointReason, SessionCommand, SessionConfig, SessionEventWire, SessionHandle,
45    SessionLifecycle, SessionReloadRecord, SessionSetting,
46};
47
48/// Env: path of the reload-state file handed to the new image.
49pub const RELOAD_STATE_ENV: &str = "SYNAPS_DAEMON_RELOAD_STATE";
50/// Env: inherited `daemon.lock` fd (mandatory when `RELOAD_STATE_ENV` is set).
51pub const LOCK_FD_ENV: &str = "SYNAPS_DAEMON_LOCK_FD";
52/// Env: the subagent handle counter read at the `execv` itself, later than
53/// `ReloadState::next_subagent_id` (written after the checkpoints): a worker
54/// that ignored its stop could still dispatch in between.
55pub const SUBAGENT_ID_FLOOR_ENV: &str = "SYNAPS_DAEMON_SUBAGENT_ID_FLOOR";
56/// Windows: env naming the event the new image sets once its `daemon.lock`
57/// request is queued behind the old image (which then exits).
58pub const HANDOFF_EVENT_ENV: &str = "SYNAPS_DAEMON_HANDOFF_EVENT";
59/// Windows: env carrying the old image's pid; the new image waits for it to
60/// be gone (and its pipe instances with it) before it binds.
61pub const HANDOFF_PID_ENV: &str = "SYNAPS_DAEMON_HANDOFF_PID";
62/// Windows: how long the old image waits for the new one to queue its lock
63/// request — process start and reading the reload-state; nothing boots first.
64#[cfg(windows)]
65const HANDOFF_ARM_TIMEOUT: Duration = Duration::from_secs(15);
66/// Windows: how long the new image waits for the old one to let go.
67#[cfg(windows)]
68const HANDOFF_EXIT_TIMEOUT: Duration = Duration::from_secs(30);
69/// Env: default drain budget in seconds.
70pub const DRAIN_SECS_ENV: &str = "SYNAPS_DAEMON_RELOAD_DRAIN_SECS";
71const DEFAULT_DRAIN: Duration = Duration::from_secs(30);
72const VERSION_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
73/// Per-session `Checkpoint{Reload}` wait: `cancel_turn` (the drain + its
74/// bounded save) + stopping the session's workers + the checkpoint's own
75/// bounded save + 1 s margin.
76const CHECKPOINT_BUDGET: Duration =
77    Duration::from_secs(crate::session::budgets::RELOAD_CHECKPOINT_BUDGET_SECS);
78/// What clients are told to wait before reconnecting.
79pub const RETRY_AFTER_MS: u64 = 500;
80
81#[derive(Debug, Clone, Default)]
82pub struct ReloadRequest {
83    pub now: bool,
84    pub drain_secs: Option<u64>,
85    pub exe: Option<PathBuf>,
86}
87
88impl ReloadRequest {
89    pub fn from_frame(frame: &ClientFrame) -> Option<Self> {
90        match frame {
91            ClientFrame::Reload { now, drain_secs, exe } => {
92                Some(Self { now: *now, drain_secs: *drain_secs, exe: exe.clone() })
93            }
94            _ => None,
95        }
96    }
97}
98
99#[derive(Debug)]
100pub enum ReloadError {
101    /// Refused before anything was disturbed (`RefuseReason::ReloadRefused`).
102    Refused(String),
103    /// `execv` returned: the old image continues; sessions are checkpointed.
104    ExecFailed(String),
105}
106
107impl std::fmt::Display for ReloadError {
108    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109        match self {
110            Self::Refused(w) => write!(f, "reload refused: {w}"),
111            Self::ExecFailed(e) => write!(f, "reload exec failed: {e}"),
112        }
113    }
114}
115
116/// `<exe> daemon --print-version` output.
117#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
118pub struct PrintVersion {
119    pub binary_version: String,
120    pub protocol_version: u32,
121}
122
123impl PrintVersion {
124    pub fn current() -> Self {
125        Self {
126            binary_version: crate::session::wire::binary_version(),
127            protocol_version: PROTOCOL_VERSION,
128        }
129    }
130}
131
132/// One recorded session.
133#[derive(Debug, Clone, Serialize, Deserialize)]
134pub struct ReloadSession {
135    pub id: String,
136    pub journal_id: String,
137    pub config: SessionConfig,
138    pub keep_warm: bool,
139    pub lifecycle: SessionLifecycle,
140    /// Non-persisted knobs re-`Set` after create (incl. `/system`).
141    #[serde(default)]
142    pub settings_replay: Vec<SessionSetting>,
143    /// Per-session `/keepwarm` override (task #444), re-applied after create.
144    #[serde(default)]
145    pub cache_keepwarm: Option<String>,
146    pub input_owner_kind: Option<crate::session::ClientKind>,
147    /// Workers the checkpoint stopped (or that still owed the parent a
148    /// completion wake), handed to the new image (`AdoptWorkers`).
149    #[serde(default)]
150    pub workers: Vec<crate::runtime::subagent::CarriedWorker>,
151    /// The session's exact tool activations, restored into the new image's
152    /// runtime (re-validated there; see `SessionToolSet::restore_carried`).
153    /// Empty from an older image or a checkpoint that did not confirm.
154    #[serde(default)]
155    pub tool_activations: Vec<crate::tools::activation::CarriedActivation>,
156    /// An owned session's owner token (`SessionConfig::owned` never rides
157    /// the config): the next image restores it owned and unbound, waiting
158    /// the re-bind grace for its owner. Never the forwarded secrets: the
159    /// owner sends them again when it re-binds.
160    #[serde(default, skip_serializing_if = "Option::is_none")]
161    pub owner_token: Option<crate::session::OwnerToken>,
162}
163
164/// `registry_dir()/daemon[-P].reload.json` body.
165#[derive(Debug, Clone, Serialize, Deserialize)]
166pub struct ReloadState {
167    pub generation: u64,
168    pub written_at: chrono::DateTime<chrono::Utc>,
169    pub sessions: Vec<ReloadSession>,
170    pub expected_clients: usize,
171    /// The subagent handle counter (`sa_N`) at the exec. The new image never
172    /// hands out a lower id, so a model's remembered handles keep meaning
173    /// the worker they named. `0` (older reload-state): derived from the
174    /// carried workers only.
175    #[serde(default)]
176    pub next_subagent_id: u64,
177}
178
179impl ReloadState {
180    /// The first handle number the new image may use: past the old image's
181    /// counter and every carried worker id.
182    pub fn subagent_id_floor(&self) -> u64 {
183        let carried = self
184            .sessions
185            .iter()
186            .flat_map(|s| &s.workers)
187            .filter_map(|w| crate::runtime::subagent::handle_number(&w.id))
188            .max()
189            .map_or(0, |n| n + 1);
190        self.next_subagent_id.max(carried)
191    }
192}
193
194pub fn reload_state_path(paths: &DaemonPaths) -> PathBuf {
195    let stem = paths
196        .json
197        .file_stem()
198        .map(|s| s.to_string_lossy().into_owned())
199        .unwrap_or_else(|| "daemon".into());
200    paths.dir.join(format!("{stem}.reload.json"))
201}
202
203/// The executable to re-exec: argv[0] canonicalised (NOT `/proc/self/exe`,
204/// which reads "(deleted)" after an in-place rebuild); `current_exe` only
205/// as a fallback when argv[0] is not resolvable (e.g. bare name on PATH).
206pub fn resolve_exe() -> PathBuf {
207    if let Some(a0) = std::env::args_os().next() {
208        let p = PathBuf::from(&a0);
209        if p.components().count() > 1 {
210            if let Ok(c) = p.canonicalize() {
211                return c;
212            }
213        }
214    }
215    std::env::current_exe().unwrap_or_else(|_| PathBuf::from("synaps"))
216}
217
218fn semver_tuple(s: &str) -> Option<(u64, u64, u64)> {
219    let core = s.split(['-', '+']).next()?;
220    let mut it = core.split('.').map(|p| p.parse::<u64>().ok());
221    Some((it.next()??, it.next()??, it.next()??))
222}
223
224/// Directional gate (§2.1): new protocol ≥ ours (clients must still be able
225/// to reconnect) and new binary ≥ ours (newer OR EQUAL; same-version
226/// rebuilds are the common case).
227pub fn version_gate(current: &PrintVersion, new: &PrintVersion) -> Result<(), String> {
228    if new.protocol_version < current.protocol_version {
229        return Err(format!(
230            "new binary speaks protocol {} < {}: reconnecting clients would be refused",
231            new.protocol_version, current.protocol_version
232        ));
233    }
234    match (semver_tuple(&current.binary_version), semver_tuple(&new.binary_version)) {
235        (Some(c), Some(n)) if n < c => Err(format!(
236            "new binary {} is older than the running {} (reload is newer-or-equal only)",
237            new.binary_version, current.binary_version
238        )),
239        (Some(_), Some(_)) => Ok(()),
240        _ => Err(format!(
241            "cannot compare versions {:?} vs {:?}",
242            current.binary_version, new.binary_version
243        )),
244    }
245}
246
247/// M8: `--exe` must be something we would trust to `execv` in our own pid:
248/// an existing regular file (canonicalised), executable, owned by our uid
249/// (or root — package-installed binaries), not group/world-writable.
250/// Returns the canonical path. Same-uid 0600 socket → not a privilege
251/// boundary; this is a footgun guard, not a sandbox.
252#[cfg(unix)]
253pub fn validate_exe(exe: &Path) -> Result<PathBuf, String> {
254    use std::os::unix::fs::{MetadataExt, PermissionsExt};
255    let canon = exe
256        .canonicalize()
257        .map_err(|e| format!("--exe {}: {e}", exe.display()))?;
258    let md = std::fs::metadata(&canon).map_err(|e| format!("--exe {}: {e}", canon.display()))?;
259    if !md.is_file() {
260        return Err(format!("--exe {}: not a regular file", canon.display()));
261    }
262    let mode = md.permissions().mode();
263    if mode & 0o111 == 0 {
264        return Err(format!("--exe {}: not executable", canon.display()));
265    }
266    // SAFETY: getuid has no preconditions.
267    let uid = unsafe { libc::getuid() };
268    if md.uid() != uid && md.uid() != 0 {
269        return Err(format!(
270            "--exe {}: owned by uid {} (expected {} or root)",
271            canon.display(),
272            md.uid(),
273            uid
274        ));
275    }
276    if mode & 0o022 != 0 {
277        return Err(format!(
278            "--exe {}: group/world-writable (mode {:o})",
279            canon.display(),
280            mode & 0o777
281        ));
282    }
283    Ok(canon)
284}
285
286#[cfg(not(unix))]
287pub fn validate_exe(exe: &Path) -> Result<PathBuf, String> {
288    exe.canonicalize().map_err(|e| format!("--exe {}: {e}", exe.display()))
289}
290
291/// Run `<exe> daemon --print-version` (5 s) and parse it.
292pub async fn probe_version(exe: &Path) -> Result<PrintVersion, String> {
293    let out = tokio::time::timeout(
294        VERSION_PROBE_TIMEOUT,
295        tokio::process::Command::new(exe)
296            .args(["daemon", "--print-version"])
297            .env("SYNAPS_DAEMON", "1")
298            .stdin(std::process::Stdio::null())
299            .stderr(std::process::Stdio::piped())
300            .stdout(std::process::Stdio::piped())
301            .output(),
302    )
303    .await
304    .map_err(|_| format!("{} --print-version timed out", exe.display()))?
305    .map_err(|e| format!("cannot run {}: {e}", exe.display()))?;
306    if !out.status.success() {
307        return Err(format!(
308            "{} --print-version exited {:?}: {}",
309            exe.display(),
310            out.status.code(),
311            String::from_utf8_lossy(&out.stderr).trim()
312        ));
313    }
314    serde_json::from_slice(&out.stdout)
315        .map_err(|e| format!("{} --print-version: unparsable output ({e})", exe.display()))
316}
317
318/// `Checkpoint{Reload}` → the actor's `SessionReloadRecord` (`None` when
319/// the checkpoint did not confirm in budget).
320async fn checkpoint_one(handle: SessionHandle) -> Option<SessionReloadRecord> {
321    let mut rx = handle.subscribe();
322    if handle
323        .send(SessionCommand::Checkpoint { reason: CheckpointReason::Reload })
324        .await
325        .is_err()
326    {
327        return None;
328    }
329    let wait = async {
330        loop {
331            match rx.recv().await {
332                Ok(env) => {
333                    if let SessionEventWire::QueryResult { id, value } = env.event {
334                        if id == CHECKPOINT_QUERY_ID {
335                            return serde_json::from_value(value["record"].clone()).ok();
336                        }
337                    }
338                }
339                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
340                Err(tokio::sync::broadcast::error::RecvError::Closed) => return None,
341            }
342        }
343    };
344    tokio::time::timeout(CHECKPOINT_BUDGET, wait).await.unwrap_or(None)
345}
346
347/// The session as it IS: the actor's record when the checkpoint confirmed,
348/// else what the handle alone knows (journal, cwd, keep-warm, lifecycle,
349/// current model from the published view, the `/keepwarm` override).
350fn record(handle: &SessionHandle, rec: Option<SessionReloadRecord>) -> ReloadSession {
351    let meta = handle.meta();
352    let journal_id = handle.journal_id();
353    if rec.is_none() {
354        tracing::warn!(
355            session = %handle.id,
356            "reload: no checkpoint reply in time — this session's workers are not carried: \
357             no archive is guaranteed and their parent will not be woken"
358        );
359    }
360    let (config, keep_warm, lifecycle, settings_replay, cache_keepwarm, model, workers, tool_activations) = match rec {
361        Some(r) => (
362            r.config,
363            r.keep_warm,
364            r.lifecycle,
365            r.settings_replay,
366            r.cache_keepwarm,
367            r.model,
368            r.workers,
369            r.tool_activations,
370        ),
371        None => (
372            SessionConfig {
373                cwd: meta.cwd.clone(),
374                env: None,
375            env_stripped: Vec::new(),
376                ..SessionConfig::default()
377            },
378            // The pin is on the handle too, so it survives a reload whose
379            // checkpoint did not confirm.
380            handle.pinned(),
381            handle.lifecycle(),
382            Vec::new(),
383            // Persisted independently of the checkpoint: an explicit
384            // `/keepwarm` choice must survive a reload that timed out.
385            handle.cache_keepwarm_override(),
386            handle.view().model.clone(),
387            // No checkpoint reply: whatever workers ran are lost to the exec
388            // without a record (the actor never answered).
389            Vec::new(),
390            // Nor are its tool activations known.
391            Vec::new(),
392        ),
393    };
394    ReloadSession {
395        id: meta.id.as_str().to_string(),
396        journal_id: journal_id.clone(),
397        config: SessionConfig {
398            continue_session: Some(Some(journal_id)),
399            model_override: Some(model),
400            keep_warm,
401            // The journal carries the CURRENT name (saveas after create);
402            // re-applying the create-time `--name` would undo a rename.
403            name: None,
404            ..config
405        },
406        keep_warm,
407        lifecycle,
408        settings_replay,
409        cache_keepwarm,
410        input_owner_kind: None,
411        workers,
412        tool_activations,
413        owner_token: handle.owner_token().cloned(),
414    }
415}
416
417/// Steps 1–4 (§2.8): gate, drain, checkpoint, reload-state, announce.
418/// Nothing irreversible has happened when this returns `Ok`; the caller
419/// (the requesting connection) sends its own `Bye{Reloading}`, flushes,
420/// then calls [`exec`].
421pub struct Prepared {
422    pub exe: PathBuf,
423    pub rs_path: PathBuf,
424    pub generation: u64,
425}
426
427pub async fn prepare(
428    state: &Arc<DaemonState>,
429    paths: &DaemonPaths,
430    req: ReloadRequest,
431) -> Result<Prepared, ReloadError> {
432    // 1. version gate — before anything is disturbed.
433    // The recorded exe from this image's own record (not daemon.json, which
434    // can be lost while the daemon runs).
435    let exe = req.exe.clone().unwrap_or_else(|| state.daemon_info().exe.unwrap_or_else(resolve_exe));
436    let exe = validate_exe(&exe).map_err(ReloadError::Refused)?;
437    let new = probe_version(&exe).await.map_err(ReloadError::Refused)?;
438    version_gate(&PrintVersion::current(), &new).map_err(ReloadError::Refused)?;
439    if state.reloading.swap(true, Ordering::SeqCst) {
440        return Err(ReloadError::Refused("a reload is already in progress".into()));
441    }
442    let generation = state.generation + 1;
443    tracing::info!(exe = %exe.display(), new = %new.binary_version, generation, "daemon: reload accepted");
444
445    // 2. drain (refuse new work), then checkpoint every session concurrently.
446    let drain = if req.now {
447        Duration::ZERO
448    } else {
449        req.drain_secs
450            .map(Duration::from_secs)
451            .or_else(|| std::env::var(DRAIN_SECS_ENV).ok()?.parse().ok().map(Duration::from_secs))
452            .unwrap_or(DEFAULT_DRAIN)
453    };
454    let t0 = std::time::Instant::now();
455    while t0.elapsed() < drain {
456        let mut all_idle = true;
457        for h in state.live_sessions() {
458            if !super::session_is_idle(&h).await {
459                all_idle = false;
460                break;
461            }
462        }
463        if all_idle {
464            break;
465        }
466        tokio::time::sleep(Duration::from_millis(200)).await;
467    }
468    let handles = state.live_sessions();
469    let results = futures::future::join_all(handles.iter().cloned().map(checkpoint_one)).await;
470    let mut sessions = Vec::with_capacity(handles.len());
471    for (h, rec) in handles.iter().zip(results) {
472        if rec.is_none() {
473            tracing::warn!(session = %h.id, "reload: checkpoint did not confirm; recording from the handle");
474        }
475        sessions.push(record(h, rec));
476    }
477
478    // 3. reload-state (0600, atomic).
479    let rs = ReloadState {
480        generation,
481        written_at: chrono::Utc::now(),
482        sessions,
483        expected_clients: state.connections.load(Ordering::SeqCst),
484        // After every checkpoint: no session can launch a worker any more.
485        next_subagent_id: crate::tools::peek_next_subagent_id(),
486    };
487    let rs_path = reload_state_path(paths);
488    // If either write fails we early-return WITHOUT reaching exec() (which is the
489    // only other place `reloading` is cleared), so reset the drain flag here —
490    // otherwise a failed reload-state write bricks the daemon (it refuses every
491    // new Create/Submit/Compact until restart).
492    let written = serde_json::to_vec_pretty(&rs)
493        .map_err(|e| e.to_string())
494        .and_then(|body| registry::write_private_atomic(&rs_path, &body).map_err(|e| e.to_string()));
495    if let Err(e) = written {
496        // No exec will follow: every checkpointed session lifts its fence
497        // and hears about the workers its checkpoint stopped.
498        abort_checkpoints(state).await;
499        state.reloading.store(false, Ordering::SeqCst);
500        return Err(ReloadError::ExecFailed(e));
501    }
502
503    // 4. announce — every OTHER conn selects on `reload_announce` and sends
504    //    Event(Reloading) + Bye{Reloading}; give their writers ≤ 1 s.
505    state.reload_generation.store(generation, Ordering::SeqCst);
506    state.fire_announce();
507    let t1 = std::time::Instant::now();
508    while state.connections.load(Ordering::SeqCst) > 1 && t1.elapsed() < Duration::from_secs(crate::session::budgets::RELOAD_ANNOUNCE_WAIT_SECS) {
509        tokio::time::sleep(Duration::from_millis(20)).await;
510    }
511    Ok(Prepared { exe, rs_path, generation })
512}
513
514/// Step 5: rewrite `daemon.json` (generation+1, same pid), stop sidecars,
515/// hand the flock to the next image and `execv`. Returns only on failure —
516/// then `daemon.json`/reload-state are restored, extension discovery is
517/// re-spawned (the sidecars were stopped for the exec — M4: no zombie
518/// daemon without hooks), and the old image keeps serving (sessions
519/// checkpointed but alive; clients that got `Bye` reconnect to THIS image
520/// — a `reconnect_of.generation` mismatch is tolerated, the daemon answers
521/// with its own generation).
522pub async fn exec(state: &Arc<DaemonState>, p: Prepared) -> ReloadError {
523    // From this image's own record, not daemon.json (which can be lost while
524    // the daemon runs). The record is held and written under the metadata
525    // lock, so a heal in the hand-off window writes the same.
526    // `previous` stays untouched: a failed exec puts it back whole.
527    let previous = state.daemon_info();
528    let prev_generation = previous.generation;
529    let info = registry::DaemonInfo {
530        generation: p.generation,
531        started_at: chrono::Utc::now(),
532        exe: Some(p.exe.clone()),
533        ..previous.clone()
534    };
535    if let Err(e) = state.meta.publish(info) {
536        tracing::warn!(path = %state.paths.json.display(), error = %e, "daemon: daemon.json write failed at reload");
537    }
538    let _ = tokio::time::timeout(
539        Duration::from_secs(crate::session::budgets::RELOAD_EXT_SHUTDOWN_SECS),
540        async { state.host.ext_manager().write().await.shutdown_all().await },
541    )
542    .await;
543
544    #[cfg(unix)]
545    let err = exec_self(&p.exe, &p.rs_path, state);
546    #[cfg(windows)]
547    let err = exec_successor(&p.exe, &p.rs_path, state).await;
548    tracing::error!(error = %err, "daemon: reload exec failed; continuing on the old image");
549    // The record before this attempt, whole: exe and started_at too, so the
550    // next default reload uses the old exe, not the one that failed.
551    if let Err(e) = state.meta.publish(previous) {
552        tracing::warn!(
553            path = %state.paths.json.display(),
554            error = %e,
555            "daemon: daemon.json write failed after the failed reload exec"
556        );
557    }
558    let _ = std::fs::remove_file(&p.rs_path);
559    state.reload_generation.store(prev_generation, Ordering::SeqCst);
560    state.reset_announce();
561    // Sidecars were shut down for the exec: bring them back (bounded like
562    // `run_foreground`'s first discovery) so the surviving image has hooks.
563    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
564    let _loader = crate::extensions::loader::spawn_discover_and_load(
565        Arc::clone(state.host.ext_manager()),
566        tx,
567        None,
568    );
569    let wait = async {
570        while let Some(ev) = rx.recv().await {
571            if let crate::extensions::loader::ExtensionLoaderEvent::Finished { loaded, failed } = ev {
572                tracing::info!(loaded = loaded.len(), failed = failed.len(), "daemon: extensions re-discovered after failed reload");
573                break;
574            }
575        }
576    };
577    if tokio::time::timeout(
578        Duration::from_secs(crate::session::budgets::RELOAD_EXT_REDISCOVERY_SECS),
579        wait,
580    )
581    .await
582    .is_err()
583    {
584        tracing::warn!("daemon: extension re-discovery still running; resuming the sessions");
585    }
586    // The checkpoint fenced every session and stopped its workers for an
587    // exec that did not happen: their parents are still here and still
588    // waiting — lift the fences and tell them.
589    abort_checkpoints(state).await;
590    state.reloading.store(false, Ordering::SeqCst);
591    ReloadError::ExecFailed(err)
592}
593
594#[cfg(unix)]
595fn exec_self(exe: &Path, rs_path: &Path, state: &DaemonState) -> String {
596    use std::os::unix::process::CommandExt;
597    let Some(lock_fd) = state.lock_fd.lock().unwrap_or_else(|e| e.into_inner()).as_ref().map(|l| l.raw_fd()) else {
598        return "no lock fd to inherit".into();
599    };
600    // Clear FD_CLOEXEC on the lock so the new image adopts the SAME flock
601    // (liveness oracle never flickers). The listener keeps CLOEXEC: it is
602    // closed at exec and the new image rebinds the path.
603    // SAFETY: fcntl on an fd we own.
604    unsafe {
605        let flags = libc::fcntl(lock_fd, libc::F_GETFD);
606        if flags >= 0 {
607            libc::fcntl(lock_fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC);
608        }
609    }
610    let mut args: Vec<std::ffi::OsString> = std::env::args_os().skip(1).collect();
611    if args.is_empty() || args.iter().all(|a| a != "daemon") {
612        args = vec!["daemon".into(), "--foreground".into()];
613    }
614    let mut cmd = std::process::Command::new(exe);
615    cmd.args(&args)
616        .env("SYNAPS_DAEMON", "1")
617        .env(LOCK_FD_ENV, lock_fd.to_string())
618        .env(RELOAD_STATE_ENV, rs_path)
619        .env(SUBAGENT_ID_FLOOR_ENV, crate::tools::peek_next_subagent_id().to_string());
620    let e = cmd.exec();
621    // Only reached on failure: restore CLOEXEC.
622    // SAFETY: as above.
623    unsafe {
624        let flags = libc::fcntl(lock_fd, libc::F_GETFD);
625        if flags >= 0 {
626            libc::fcntl(lock_fd, libc::F_SETFD, flags | libc::FD_CLOEXEC);
627        }
628    }
629    e.to_string()
630}
631
632/// Windows `execv`: start the new image with the reload-state and a
633/// hand-off event, wait for it to queue its `daemon.lock` request behind
634/// ours, then exit — no destructors (they would unlink the `daemon.json`
635/// that now names the next generation), no session `End` (the new image
636/// rehydrates them): what the exec discards on Unix. Returns only on
637/// failure, with this image intact.
638#[cfg(windows)]
639async fn exec_successor(exe: &Path, rs_path: &Path, state: &DaemonState) -> String {
640    use super::win;
641    let armed = match win::Event::create_unique("handoff") {
642        Ok(e) => e,
643        Err(e) => return format!("hand-off event: {e}"),
644    };
645    let mut args: Vec<std::ffi::OsString> = std::env::args_os().skip(1).collect();
646    if args.is_empty() || args.iter().all(|a| a != "daemon") {
647        args = vec!["daemon".into(), "--foreground".into()];
648    }
649    let mut env = crate::process_scope::inherited_env();
650    env.retain(|(k, _)| k != super::READY_EVENT_ENV);
651    crate::process_scope::set_env(&mut env, "SYNAPS_DAEMON", "1");
652    crate::process_scope::set_env(&mut env, RELOAD_STATE_ENV, rs_path.as_os_str());
653    crate::process_scope::set_env(&mut env, HANDOFF_EVENT_ENV, armed.name());
654    crate::process_scope::set_env(&mut env, HANDOFF_PID_ENV, std::process::id().to_string());
655    crate::process_scope::set_env(&mut env, SUBAGENT_ID_FLOOR_ENV, crate::tools::peek_next_subagent_id().to_string());
656    let child = match win::spawn_detached(exe, &args, &env) {
657        Ok(c) => c,
658        Err(e) => return format!("cannot start {}: {e}", exe.display()),
659    };
660    let waited = tokio::task::spawn_blocking(move || {
661        let outcome = win::wait_event_or_exit(&armed, &child, HANDOFF_ARM_TIMEOUT);
662        (child, outcome)
663    })
664    .await;
665    let (child, outcome) = match waited {
666        Ok(r) => r,
667        Err(e) => return format!("hand-off wait: {e}"),
668    };
669    match outcome {
670        win::Wait::Signalled => {
671            tracing::info!(pid = child.id(), "daemon: reload hand-off armed; this image exits");
672            state.host.flush_logs();
673            std::process::exit(0);
674        }
675        win::Wait::Exited => {
676            let mut child = child;
677            let code = child.wait(Duration::ZERO);
678            let tail = win::stderr_tail(child.take_stderr(), 64 * 1024, Duration::from_millis(500));
679            let why = tail
680                .lines()
681                .map(str::trim)
682                .rfind(|l| !l.is_empty())
683                .map(|l| l.trim_start_matches("synaps daemon: ").to_string())
684                .unwrap_or_else(|| "no output".into());
685            format!(
686                "new image exited (code {}) before taking over: {why}",
687                code.map_or_else(|| "?".to_string(), |c| c.to_string())
688            )
689        }
690        win::Wait::TimedOut => {
691            child.kill();
692            let _ = child.wait(Duration::from_secs(5));
693            format!("new image did not take over within {HANDOFF_ARM_TIMEOUT:?}")
694        }
695    }
696}
697
698/// What a new image adopts: the held lock, the reload-state and its path.
699type Adopted = (DaemonLock, ReloadState, PathBuf);
700
701/// Windows: the hand-off taken at the top of `run_foreground`, held here
702/// for [`adopt_from_env`].
703#[cfg(windows)]
704static TAKEN_OVER: std::sync::Mutex<Option<Adopted>> = std::sync::Mutex::new(None);
705
706/// Windows: take `daemon.lock` over from the previous image now, before
707/// anything boots (the previous image exits on our signal, as the exec
708/// would have replaced it). A no-op unless this is a reload.
709#[cfg(windows)]
710pub fn take_over_from_env(paths: &DaemonPaths) -> anyhow::Result<()> {
711    if let Some(adopted) = handoff_from_env(paths)? {
712        *TAKEN_OVER.lock().unwrap_or_else(|e| e.into_inner()) = Some(adopted);
713    }
714    Ok(())
715}
716
717/// New-image side (`Daemon::start`), Windows: what [`take_over_from_env`]
718/// took, or the hand-off done now.
719#[cfg(windows)]
720pub fn adopt_from_env(paths: &DaemonPaths) -> anyhow::Result<Option<Adopted>> {
721    if let Some(adopted) = TAKEN_OVER.lock().unwrap_or_else(|e| e.into_inner()).take() {
722        return Ok(Some(adopted));
723    }
724    handoff_from_env(paths)
725}
726
727/// Windows hand-off, new-image side: parse the reload-state first (a bad
728/// one fails while the old image can still carry on), queue `daemon.lock`
729/// behind the old image and set its hand-off event, take the grant when it
730/// exits, then wait for its process to be gone. The event is mandatory —
731/// without it the old image would never let go (refuse, as Unix refuses a
732/// reload-state without its lock fd).
733#[cfg(windows)]
734fn handoff_from_env(paths: &DaemonPaths) -> anyhow::Result<Option<Adopted>> {
735    let Some(rs_path) = std::env::var_os(RELOAD_STATE_ENV) else {
736        return Ok(None);
737    };
738    let rs_path = PathBuf::from(rs_path);
739    std::env::remove_var(RELOAD_STATE_ENV);
740    let event = std::env::var(HANDOFF_EVENT_ENV).ok();
741    std::env::remove_var(HANDOFF_EVENT_ENV);
742    let old_pid = std::env::var(HANDOFF_PID_ENV).ok().and_then(|s| s.parse::<u32>().ok());
743    std::env::remove_var(HANDOFF_PID_ENV);
744    let id_floor = std::env::var(SUBAGENT_ID_FLOOR_ENV).ok().and_then(|s| s.parse::<u64>().ok());
745    std::env::remove_var(SUBAGENT_ID_FLOOR_ENV);
746    let event = event.ok_or_else(|| {
747        anyhow::anyhow!("{RELOAD_STATE_ENV} set without {HANDOFF_EVENT_ENV}: refusing to start (the lock is the liveness oracle)")
748    })?;
749    let body = std::fs::read_to_string(&rs_path)?;
750    let mut rs: ReloadState = serde_json::from_str(&body)?;
751    if let Some(floor) = id_floor {
752        rs.next_subagent_id = rs.next_subagent_id.max(floor);
753    }
754    // Opened before the signal, so its pid cannot be reused under us.
755    let old = old_pid.and_then(super::win::Process::open);
756    let mut signal_err = None;
757    let lock = DaemonLock::acquire_handoff(paths, HANDOFF_EXIT_TIMEOUT, || {
758        signal_err = super::win::signal_event(&event).err();
759    })
760    .map_err(|e| match &signal_err {
761        Some(s) => anyhow::anyhow!("taking daemon.lock over: {e} (the previous image was never told: {s})"),
762        None => anyhow::anyhow!("taking daemon.lock over from the previous image: {e}"),
763    })?;
764    if let Some(old) = old {
765        if !old.wait(Duration::from_secs(10)) {
766            tracing::warn!(pid = old_pid, "daemon: the previous image still runs after releasing daemon.lock");
767        }
768    }
769    tracing::info!(generation = rs.generation, "daemon: took daemon.lock over from the previous image");
770    // The previous image read our stderr; it is gone.
771    super::win::detach_stdio();
772    Ok(Some((lock, rs, rs_path)))
773}
774
775/// New-image side (`Daemon::start`): when `RELOAD_STATE_ENV` is set, the
776/// lock is ADOPTED from `LOCK_FD_ENV` (mandatory — refuse to start without
777/// it, risk §6.8) and the recorded generation is returned.
778#[cfg(unix)]
779pub fn adopt_from_env(_paths: &DaemonPaths) -> anyhow::Result<Option<Adopted>> {
780    let Some(rs_path) = std::env::var_os(RELOAD_STATE_ENV) else {
781        return Ok(None);
782    };
783    let rs_path = PathBuf::from(rs_path);
784    std::env::remove_var(RELOAD_STATE_ENV);
785    let fd: i32 = std::env::var(LOCK_FD_ENV)
786        .ok()
787        .and_then(|s| s.parse().ok())
788        .ok_or_else(|| anyhow::anyhow!("{RELOAD_STATE_ENV} set without {LOCK_FD_ENV}: refusing to start (the flock is the liveness oracle)"))?;
789    std::env::remove_var(LOCK_FD_ENV);
790    let id_floor = std::env::var(SUBAGENT_ID_FLOOR_ENV).ok().and_then(|s| s.parse::<u64>().ok());
791    std::env::remove_var(SUBAGENT_ID_FLOOR_ENV);
792    let lock = DaemonLock::adopt(fd)?;
793    let body = std::fs::read_to_string(&rs_path)?;
794    let mut rs: ReloadState = serde_json::from_str(&body)?;
795    if let Some(floor) = id_floor {
796        rs.next_subagent_id = rs.next_subagent_id.max(floor);
797    }
798    Ok(Some((lock, rs, rs_path)))
799}
800
801/// Rehydrate every recorded session BEFORE accepting. A session that
802/// fails to come back is logged and skipped (its journal is on disk;
803/// `--continue` brings it back).
804///
805/// A session whose journal was never written (no turn yet — `save` skips
806/// an empty conversation) cannot be continued; it is recreated fresh under
807/// a new id and aliased from the old one.
808/// F23: detect "is live in another process" in the error string — this is
809/// the `SessionLockError::Held` case, where an in-process runtime holds
810/// the journal lock. Instead of aliasing to an empty impostor, register a
811/// Parked placeholder under the SAME id.
812pub(crate) fn is_lock_held_error(e: &str) -> bool {
813    e.contains("is live in another process")
814}
815
816/// Extract `"pid N (kind)"` from the lock-held error message.
817pub(crate) fn parse_holder_from_error(e: &str) -> String {
818    // "session X is live in another process (pid N, kind) — …"
819    if let Some(start) = e.find("(pid ") {
820        let after = &e[start + 1..]; // "pid N, kind) — …"
821        if let Some(end) = after.find(')') {
822            return after[..end].to_string();
823        }
824    }
825    "unknown holder".to_string()
826}
827
828/// No exec after the checkpoints (it failed, or the reload stopped before
829/// it): every live session lifts its reload fence and is handed back the
830/// workers its checkpoint stopped (`ReloadAborted`, a no-op for a session
831/// that never checkpointed). The actor re-announces from its own record,
832/// so a checkpoint that answered too late for the reload record is covered.
833async fn abort_checkpoints(state: &Arc<DaemonState>) {
834    let budget = Duration::from_secs(crate::session::budgets::RELOAD_ABORT_SEND_SECS);
835    let sends = state
836        .live_sessions()
837        .into_iter()
838        .map(|h| async move { send_reload_aborted(&h, budget).await });
839    futures::future::join_all(sends).await;
840}
841
842/// `ReloadAborted` must arrive: dropped, the session would wait out the
843/// fence's backstop. `SessionHandle::send` never waits for queue room, so
844/// retry a full queue (the actor drains it) for up to `budget`.
845pub(crate) async fn send_reload_aborted(h: &SessionHandle, budget: Duration) -> bool {
846    use crate::session::TransportError;
847    let deadline = tokio::time::Instant::now() + budget;
848    loop {
849        match h.send(SessionCommand::ReloadAborted).await {
850            Ok(()) => return true,
851            Err(TransportError::Backpressure) if tokio::time::Instant::now() < deadline => {
852                tokio::time::sleep(Duration::from_millis(20)).await;
853            }
854            Err(e) => {
855                tracing::warn!(
856                    session = %h.id,
857                    error = %e,
858                    "reload: could not tell the session the exec failed; its reload fence lifts at the backstop"
859                );
860                return false;
861            }
862        }
863    }
864}
865
866/// The config a session is restored with: as saved, except that the
867/// restore never waits on `on_session_start`. Each restored session runs
868/// the hook in the background (the deferred path quick-start sessions
869/// use), so the restore of N sessions does not take N hook round trips
870/// while the socket waits. A session restored Parked runs it only once it
871/// is unparked.
872///
873/// A folder that is no longer there is dropped ([`restorable_folder`]), as
874/// the pinned restore drops it (`pins::restore_one`): the chat `chat` then
875/// runs in the daemon's folder and its project. Kept, the dead path gave
876/// the chat no project at all: discovery fails on a path it cannot resolve,
877/// the memory tools answer "host project scope unavailable" and the next
878/// rollover cannot seal.
879fn restore_config(saved: &SessionConfig, lifecycle: SessionLifecycle, chat: &str) -> SessionConfig {
880    SessionConfig {
881        await_extensions: false,
882        restore_parked: matches!(lifecycle, SessionLifecycle::Parked | SessionLifecycle::Parking),
883        cwd: restorable_folder(saved.cwd.clone(), chat),
884        ..saved.clone()
885    }
886}
887
888/// The folder a chat is restored in, by a reload, as a pinned session or by
889/// an unpark (`session/actor.rs`):
890/// `folder` as saved, or `None` when it is gone, and the chat then runs in
891/// the daemon's folder and that folder's project.
892///
893/// Gone means that the path does not exist, is not a directory, or runs
894/// through something that is not a directory (a file where one of its
895/// parents was), which no folder can be under. A folder that cannot be
896/// looked at for a reason that can pass (no permission, a mount that is
897/// away for a moment, an I/O error) is kept: the chat stays in its
898/// own project, where its tools fail until the folder is back, rather than
899/// being moved to another project's files and memory without a word. Every
900/// drop is logged with the folder and the chat. The lines name no host: an
901/// unpark in a host that is not the daemon writes them too.
902pub(crate) fn restorable_folder(folder: Option<PathBuf>, chat: &str) -> Option<PathBuf> {
903    let folder = folder?;
904    let gone = match std::fs::metadata(&folder) {
905        Ok(found) if found.is_dir() => return Some(folder),
906        Ok(_) => "it is not a directory",
907        Err(error) if error.kind() == std::io::ErrorKind::NotFound => "it does not exist",
908        Err(error) if error.kind() == std::io::ErrorKind::NotADirectory => {
909            "its path runs through something that is not a directory"
910        }
911        Err(error) => {
912            tracing::warn!(
913                session = %chat,
914                folder = %folder.display(),
915                error = %error,
916                "{}",
917                UNSEEN_NOTICE
918            );
919            return Some(folder);
920        }
921    };
922    tracing::warn!(
923        session = %chat,
924        folder = %folder.display(),
925        "{}",
926        gone_notice(gone)
927    );
928    None
929}
930
931/// What [`restorable_folder`] logs for a folder it keeps without having
932/// seen it. Like [`gone_notice`] it names no host.
933const UNSEEN_NOTICE: &str = "the chat's folder cannot be looked at; restored with it all the same";
934
935/// What [`restorable_folder`] logs for a folder it drops, `gone` being why.
936/// It names no host ("daemon:" it once began with): a reload and a pinned
937/// restore run in the daemon, an unpark in whichever host holds the chat.
938fn gone_notice(gone: &str) -> String {
939    format!(
940        "the chat's folder is gone ({gone}); it is restored in the host process's folder \
941         and that folder's project"
942    )
943}
944
945pub async fn rehydrate(state: &Arc<DaemonState>, rs: &ReloadState) {
946    // Before any session can dispatch: ids continue where the old image
947    // stopped, so `sa_1` never names two different workers.
948    crate::tools::seed_next_subagent_id(rs.subagent_id_floor());
949    for s in &rs.sessions {
950        // Owned: restored unbound, its grace armed only once this image
951        // accepts connections (`Daemon::start`), its autonomous work held
952        // until the owner re-binds.
953        let owned = s.owner_token.clone().map(|token| crate::session::OwnedConfig { token, restored: true });
954        let config = SessionConfig { owned, ..restore_config(&s.config, s.lifecycle, &s.id) };
955        let created = match state.create(config.clone()).await {
956            Ok(h) => Ok(h),
957            // F23: journal locked by another process — register a Parked
958            // placeholder under the SAME id. No alias, no fresh session.
959            Err(e) if is_lock_held_error(&e) => {
960                let holder = parse_holder_from_error(&e);
961                tracing::warn!(
962                    session = %s.id, holder = %holder,
963                    "daemon: journal locked by another process; registering Parked placeholder"
964                );
965                let sid = crate::session::SessionId::from(s.id.as_str());
966                // `config`, not `s.config`: an owned session stays owned.
967                let (handle, _task) = crate::session::handle::locked_placeholder::spawn(
968                    sid,
969                    config.clone(),
970                    holder,
971                );
972                state.insert(handle);
973                continue;
974            }
975            Err(e) if s.config.continue_session.is_some() => {
976                tracing::warn!(session = %s.id, error = %e, "daemon: journal not continuable; recreating fresh");
977                state
978                    .create(SessionConfig {
979                        continue_session: None,
980                        ..config.clone()
981                    })
982                    .await
983            }
984            Err(e) => Err(e),
985        };
986        match created {
987            Ok(h) => {
988                let new_id = h.id.as_str().to_string();
989                // First, before anything can start a turn: the same
990                // conversation continues in this image, so the grants it
991                // held come with it (re-validated by the new runtime at its
992                // first turn start). A session recreated fresh (journal not
993                // continuable) is a new conversation and inherits none.
994                let continued = h.meta().continued;
995                if continued && !s.tool_activations.is_empty() {
996                    let _ = h
997                        .send(SessionCommand::RestoreToolActivations {
998                            activations: s.tool_activations.clone(),
999                        })
1000                        .await;
1001                }
1002                if new_id != s.id {
1003                    state
1004                        .reload_aliases
1005                        .lock()
1006                        .unwrap_or_else(|e| e.into_inner())
1007                        .insert(s.id.clone(), new_id.clone());
1008                }
1009                // Non-persisted knobs (host-originated `Set`: no owner check).
1010                for (i, setting) in s.settings_replay.iter().enumerate() {
1011                    let _ = h
1012                        .send(SessionCommand::Set {
1013                            id: u64::MAX - i as u64,
1014                            setting: setting.clone(),
1015                        })
1016                        .await;
1017                }
1018                if s.keep_warm {
1019                    let _ = h.send(SessionCommand::KeepWarm { on: true }).await;
1020                }
1021                if let Some(mode) = &s.cache_keepwarm {
1022                    let _ = h
1023                        .send(SessionCommand::EngineCommand {
1024                            id: u64::MAX,
1025                            name: "keepwarm".into(),
1026                            arg: mode.clone(),
1027                        })
1028                        .await;
1029                }
1030                // Before any Park: re-register the stopped workers and wake
1031                // their parent (it was waiting on completions the exec ate).
1032                if !s.workers.is_empty() {
1033                    let _ = h
1034                        .send(SessionCommand::AdoptWorkers { workers: s.workers.clone() })
1035                        .await;
1036                }
1037                if matches!(s.lifecycle, SessionLifecycle::Parked | SessionLifecycle::Parking) {
1038                    let _ = h.send(SessionCommand::Park).await;
1039                } else {
1040                    // Last, after the pin, keep-warm and workers it weighs:
1041                    // the old image's park timer died with the exec, and a
1042                    // session nobody reconnects to gets no detach to arm one.
1043                    let _ = h.send(SessionCommand::ArmPark).await;
1044                }
1045                tracing::info!(old = %s.id, new = %new_id, lifecycle = ?s.lifecycle, "daemon: session rehydrated after reload");
1046            }
1047            Err(e) => tracing::warn!(session = %s.id, error = %e, "daemon: session could not be rehydrated"),
1048        }
1049    }
1050}
1051
1052#[cfg(test)]
1053mod tests {
1054    use super::*;
1055
1056    /// The session's tool activations ride the reload-state file; a file
1057    /// from an older image (no field) reads as none.
1058    #[test]
1059    fn reload_session_carries_tool_activations() {
1060        use crate::tools::activation::{CarriedActivation, CarriedProvenance};
1061        let carried = vec![CarriedActivation {
1062            tool_id: "builtin:subagent_start".into(),
1063            schema_digest: "cd".repeat(32),
1064            provenance: CarriedProvenance::BuiltinRuntime,
1065        }];
1066        let s = ReloadSession {
1067            id: "s".into(),
1068            journal_id: "j".into(),
1069            config: SessionConfig::default(),
1070            keep_warm: false,
1071            lifecycle: SessionLifecycle::Live,
1072            settings_replay: Vec::new(),
1073            cache_keepwarm: None,
1074            input_owner_kind: None,
1075            workers: Vec::new(),
1076            tool_activations: carried.clone(),
1077            owner_token: None,
1078        };
1079        let v = serde_json::to_value(&s).unwrap();
1080        let back: ReloadSession = serde_json::from_value(v.clone()).unwrap();
1081        assert_eq!(back.tool_activations, carried);
1082        // Never via the create config's wire form.
1083        assert!(v["config"].get("restore_tool_activations").is_none());
1084
1085        let mut older = v;
1086        older.as_object_mut().unwrap().remove("tool_activations");
1087        let back: ReloadSession = serde_json::from_value(older).unwrap();
1088        assert!(back.tool_activations.is_empty());
1089    }
1090
1091    /// Reload-state written by an older image (no workers, no counter)
1092    /// still parses: the new fields default.
1093    #[test]
1094    fn reload_state_from_an_older_image_parses() {
1095        let rs: ReloadState = serde_json::from_value(serde_json::json!({
1096            "generation": 9,
1097            "written_at": "2026-10-14T18:17:03Z",
1098            "sessions": [],
1099            "expected_clients": 1
1100        }))
1101        .unwrap();
1102        assert_eq!(rs.next_subagent_id, 0);
1103        assert_eq!(rs.subagent_id_floor(), 0);
1104    }
1105
1106    /// The new image's first handle is past the old counter AND past every
1107    /// carried id (an older writer that only carried workers still works).
1108    #[test]
1109    fn subagent_id_floor_covers_the_counter_and_every_carried_id() {
1110        let worker = |id: &str| crate::runtime::subagent::CarriedWorker {
1111            id: id.into(),
1112            subagent_id: 0,
1113            agent: String::new(),
1114            status: "cancelled".into(),
1115            archive_path: None,
1116            stopped_by_reload: true,
1117            wake_lost: false,
1118            wake: true,
1119            elapsed_secs: 0,
1120            preview: String::new(),
1121        };
1122        let session = |workers| ReloadSession {
1123            id: "s".into(),
1124            journal_id: "j".into(),
1125            config: SessionConfig::default(),
1126            keep_warm: false,
1127            lifecycle: SessionLifecycle::Live,
1128            settings_replay: Vec::new(),
1129            cache_keepwarm: None,
1130            input_owner_kind: None,
1131            workers,
1132            tool_activations: Vec::new(),
1133            owner_token: None,
1134        };
1135        let mut rs = ReloadState {
1136            generation: 10,
1137            written_at: chrono::Utc::now(),
1138            sessions: vec![session(vec![worker("sa_288"), worker("sa_286")]), session(vec![])],
1139            expected_clients: 0,
1140            next_subagent_id: 0,
1141        };
1142        assert_eq!(rs.subagent_id_floor(), 289);
1143        rs.next_subagent_id = 300;
1144        assert_eq!(rs.subagent_id_floor(), 300);
1145        // Round trip keeps both.
1146        let back: ReloadState = serde_json::from_slice(&serde_json::to_vec(&rs).unwrap()).unwrap();
1147        assert_eq!(back.subagent_id_floor(), 300);
1148        assert_eq!(back.sessions[0].workers.len(), 2);
1149    }
1150
1151    fn pv(b: &str, p: u32) -> PrintVersion {
1152        PrintVersion { binary_version: b.into(), protocol_version: p }
1153    }
1154
1155    #[test]
1156    fn gate_is_directional_newer_or_equal() {
1157        let cur = pv("0.9.0", 2);
1158        assert!(version_gate(&cur, &pv("0.9.0", 2)).is_ok(), "equal allowed");
1159        assert!(version_gate(&cur, &pv("0.9.1", 2)).is_ok());
1160        assert!(version_gate(&cur, &pv("1.0.0-rc1", 3)).is_ok());
1161        assert!(version_gate(&cur, &pv("0.8.9", 2)).is_err(), "older refused");
1162        assert!(version_gate(&cur, &pv("0.9.0", 1)).is_err(), "older protocol refused");
1163        assert!(version_gate(&cur, &pv("garbage", 2)).is_err());
1164        // Two-digit minor (daemon dev builds report 0.9.1 → 0.10.0): compared
1165        // numerically, not as strings — "10" < "9" lexically would refuse it.
1166        assert!(
1167            version_gate(&pv("0.9.1", 3), &pv("0.10.0", 3)).is_ok(),
1168            "0.10.0 is newer than 0.9.1"
1169        );
1170        assert!(
1171            version_gate(&pv("0.10.0", 3), &pv("0.9.1", 3)).is_err(),
1172            "0.9.1 is older than 0.10.0"
1173        );
1174    }
1175
1176    #[cfg(unix)]
1177    #[test]
1178    fn validate_exe_refuses_missing_dirs_non_exec_and_writable() {
1179        use std::os::unix::fs::PermissionsExt;
1180        let d = tempfile::tempdir().unwrap();
1181        assert!(validate_exe(&d.path().join("nope")).unwrap_err().contains("nope"));
1182        assert!(validate_exe(d.path()).unwrap_err().contains("not a regular file"));
1183        let f = d.path().join("bin");
1184        std::fs::write(&f, "#!/bin/sh\nexit 0\n").unwrap();
1185        std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o644)).unwrap();
1186        assert!(validate_exe(&f).unwrap_err().contains("not executable"));
1187        std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o777)).unwrap();
1188        assert!(validate_exe(&f).unwrap_err().contains("writable"));
1189        std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o755)).unwrap();
1190        assert_eq!(validate_exe(&f).unwrap(), f.canonicalize().unwrap());
1191        // A symlink resolves to its target.
1192        let l = d.path().join("link");
1193        std::os::unix::fs::symlink(&f, &l).unwrap();
1194        assert_eq!(validate_exe(&l).unwrap(), f.canonicalize().unwrap());
1195    }
1196
1197    /// A restored session never waits on `on_session_start` at create; the
1198    /// rest of its saved config is kept. One restored Parked defers the hook
1199    /// to its unpark, and that flag never goes on the wire.
1200    #[test]
1201    fn a_restore_does_not_wait_on_the_session_start_hook() {
1202        let saved = SessionConfig {
1203            model_override: Some("claude-sonnet-4-5".into()),
1204            keep_warm: true,
1205            await_extensions: true,
1206            ..Default::default()
1207        };
1208        let restored = restore_config(&saved, SessionLifecycle::Live, "chat");
1209        assert!(!restored.await_extensions);
1210        assert!(!restored.restore_parked);
1211        let as_saved = SessionConfig {
1212            await_extensions: true,
1213            ..restored
1214        };
1215        assert_eq!(
1216            serde_json::to_value(&as_saved).unwrap(),
1217            serde_json::to_value(&saved).unwrap()
1218        );
1219        for parked in [SessionLifecycle::Parked, SessionLifecycle::Parking] {
1220            let restored = restore_config(&saved, parked, "chat");
1221            assert!(restored.restore_parked, "{parked:?}");
1222            assert!(!restored.await_extensions);
1223            let wire = serde_json::to_value(&restored).unwrap();
1224            assert!(wire.get("restore_parked").is_none(), "never on the wire");
1225        }
1226    }
1227
1228    /// A chat whose folder was deleted while the daemon held it. The pinned
1229    /// restore drops such a folder (`pins::restore_one`, by the same
1230    /// [`restorable_folder`]) and the chat runs in the
1231    /// daemon's folder: a runtime without a folder of its own takes the
1232    /// process's. A reload kept the dead path. The chat's memory binding was
1233    /// then built on a path project discovery cannot resolve, so it had no
1234    /// project: memory tools answered "host project scope unavailable" and a
1235    /// rollover could not seal. Reload now drops it too.
1236    #[test]
1237    fn a_restore_drops_a_folder_that_is_gone_as_the_pinned_restore_does() {
1238        let parent = tempfile::tempdir().unwrap();
1239        let (kept, gone) = (parent.path().join("kept"), parent.path().join("gone"));
1240        std::fs::create_dir(&kept).unwrap();
1241        let saved = |folder: &Path| SessionConfig {
1242            cwd: Some(folder.to_path_buf()),
1243            ..Default::default()
1244        };
1245
1246        // What the dead path does to a chat that keeps it.
1247        let binding = crate::memory_backend::MemoryBinding::for_cwd(Some(gone.clone()));
1248        let error = binding.scope().unwrap_err().to_string();
1249        assert!(error.contains("host project scope unavailable"), "{error}");
1250        let binding = crate::memory_backend::MemoryBinding::for_cwd(Some(kept.clone()));
1251        assert!(binding.scope().is_ok());
1252
1253        for lifecycle in [SessionLifecycle::Live, SessionLifecycle::Parked] {
1254            assert_eq!(restore_config(&saved(&kept), lifecycle, "chat").cwd, Some(kept.clone()));
1255            assert_eq!(restore_config(&saved(&gone), lifecycle, "chat").cwd, None, "{lifecycle:?}");
1256        }
1257        // A file where the folder was is not a folder to run in either.
1258        std::fs::write(&gone, b"").unwrap();
1259        assert_eq!(restore_config(&saved(&gone), SessionLifecycle::Live, "chat").cwd, None);
1260        // No folder saved: nothing to drop.
1261        assert_eq!(restore_config(&SessionConfig::default(), SessionLifecycle::Live, "chat").cwd, None);
1262    }
1263
1264    /// The two lines `restorable_folder` logs do not say "daemon": an unpark
1265    /// in a host that is not the daemon goes through the same function.
1266    /// (The text is checked, not a captured log: a log captured on one test
1267    /// thread loses lines to tracing's callsite cache when another test
1268    /// reaches the same line first.)
1269    #[test]
1270    fn the_lines_about_a_chats_folder_name_no_host() {
1271        let gone = gone_notice("it does not exist");
1272        assert!(gone.starts_with("the chat's folder is gone (it does not exist); "), "{gone}");
1273        for line in [gone.as_str(), UNSEEN_NOTICE] {
1274            assert!(line.starts_with("the chat's folder "), "{line}");
1275            assert!(!line.contains("daemon"), "{line}");
1276        }
1277    }
1278
1279    /// The rule a reload and the pinned restore share. A folder is dropped,
1280    /// and the chat moved to the daemon's folder and project, only when it
1281    /// is known to be gone. `is_dir()` said "gone" for every failure, so a
1282    /// folder that could not be looked at for a moment (a mount that was
1283    /// away, a permission) moved the chat as well, without a log line.
1284    #[cfg(unix)]
1285    #[test]
1286    fn a_folder_is_dropped_only_when_it_is_absent_or_not_a_directory() {
1287        use std::os::unix::fs::PermissionsExt;
1288        let parent = tempfile::tempdir().unwrap();
1289        let at = |name: &str| parent.path().join(name);
1290        let restored = |folder: &Path| restorable_folder(Some(folder.to_path_buf()), "chat");
1291        assert_eq!(restorable_folder(None, "chat"), None);
1292
1293        // Absent, and a file: dropped.
1294        assert_eq!(restored(&at("absent")), None);
1295        std::fs::write(at("file"), b"").unwrap();
1296        assert_eq!(restored(&at("file")), None);
1297        // A path under a file (ENOTDIR, not ENOENT): no folder can be
1298        // there, so it is absent, whatever is asked for under it.
1299        let under = at("file").join("project");
1300        let not_a_directory = std::fs::metadata(&under).unwrap_err().kind();
1301        assert_eq!(not_a_directory, std::io::ErrorKind::NotADirectory);
1302        assert_eq!(restored(&under), None);
1303        assert_eq!(restored(&under.join("deeper")), None);
1304        // A link to nothing is absent too.
1305        std::os::unix::fs::symlink(at("absent"), at("dangling")).unwrap();
1306        assert_eq!(restored(&at("dangling")), None);
1307
1308        // A directory, and a link to one: kept, as saved.
1309        std::fs::create_dir(at("dir")).unwrap();
1310        assert_eq!(restored(&at("dir")), Some(at("dir")));
1311        std::os::unix::fs::symlink(at("dir"), at("link")).unwrap();
1312        assert_eq!(restored(&at("link")), Some(at("link")));
1313
1314        // A folder that is there and cannot be looked at: kept. The same
1315        // answer whether or not it exists behind the closed directory, which
1316        // is the point: nobody can tell from here.
1317        let closed = at("closed");
1318        std::fs::create_dir_all(closed.join("project")).unwrap();
1319        let mode = |mode: u32| {
1320            std::fs::set_permissions(&closed, std::fs::Permissions::from_mode(mode)).unwrap()
1321        };
1322        mode(0o000);
1323        let denied = |path: &Path| {
1324            std::fs::metadata(path).is_err_and(|e| e.kind() == std::io::ErrorKind::PermissionDenied)
1325        };
1326        // Root reads through any mode: nothing to test there.
1327        if denied(&closed.join("project")) {
1328            assert!(!closed.join("project").is_dir(), "what the old rule went by");
1329            for hidden in [closed.join("project"), closed.join("nothing")] {
1330                assert_eq!(restored(&hidden), Some(hidden.clone()));
1331            }
1332        }
1333        mode(0o700);
1334        assert_eq!(restored(&closed.join("project")), Some(closed.join("project")));
1335        assert_eq!(restored(&closed.join("nothing")), None);
1336    }
1337
1338    #[test]
1339    fn reload_state_path_follows_profile() {
1340        let d = tempfile::tempdir().unwrap();
1341        let p = registry::daemon_paths_in(d.path(), Some("work"));
1342        assert!(reload_state_path(&p).ends_with("daemon-work.reload.json"), "{:?}", reload_state_path(&p));
1343        let p = registry::daemon_paths_in(d.path(), None);
1344        assert!(reload_state_path(&p).ends_with("daemon.reload.json"));
1345    }
1346
1347    /// Task #444: a reload whose checkpoint did not confirm (`rec == None`)
1348    /// still carries the explicit `/keepwarm` choice, read from the handle,
1349    /// instead of resurrecting the config default.
1350    #[test]
1351    fn keepwarm_override_survives_an_unconfirmed_checkpoint() {
1352        use crate::core::config::CacheKeepWarm;
1353        use crate::session::handle::echo::{meta_for, view_for};
1354        let id = crate::session::SessionId("kw-reload".into());
1355        let (handle, ep) = SessionHandle::new(meta_for(&id), view_for());
1356        assert_eq!(record(&handle, None).cache_keepwarm, None, "no override yet");
1357        // What the actor writes on `/keepwarm off`.
1358        *ep.cache_keepwarm.lock().unwrap() = Some(CacheKeepWarm::Off);
1359        assert_eq!(record(&handle, None).cache_keepwarm.as_deref(), Some("off"));
1360        *ep.cache_keepwarm.lock().unwrap() = Some(CacheKeepWarm::Hours(6));
1361        assert_eq!(record(&handle, None).cache_keepwarm.as_deref(), Some("6h"));
1362    }
1363
1364    /// Likewise the pin: an unconfirmed checkpoint must not unpin a
1365    /// session (it would park, and stop holding the daemon's idle exit).
1366    #[test]
1367    fn the_pin_survives_an_unconfirmed_checkpoint() {
1368        use crate::session::handle::echo::{meta_for, view_for};
1369        let id = crate::session::SessionId("pin-reload".into());
1370        let (handle, ep) = SessionHandle::new(meta_for(&id), view_for());
1371        assert!(!record(&handle, None).keep_warm);
1372        ep.keep_warm.store(true, std::sync::atomic::Ordering::Release);
1373        let r = record(&handle, None);
1374        assert!(r.keep_warm && r.config.keep_warm);
1375    }
1376
1377    /// T-B4: a session config written before the session memory mode was
1378    /// removed still carries `memory_mode`. It is read and ignored, in the
1379    /// config alone and inside a reload state, whatever its value, and the
1380    /// rest of the config is kept.
1381    #[test]
1382    fn an_old_memory_mode_is_read_and_ignored() {
1383        for v in [
1384            serde_json::json!("capture_and_recall"),
1385            serde_json::json!("off"),
1386            serde_json::json!("recall_forever"),
1387            serde_json::json!(7),
1388            serde_json::Value::Null,
1389        ] {
1390            let cfg: SessionConfig = serde_json::from_value(serde_json::json!({
1391                "memory_mode": v,
1392                "model_override": "claude-sonnet-4-5",
1393                "keep_warm": true
1394            }))
1395            .unwrap_or_else(|e| panic!("memory_mode = {v}: {e}"));
1396            assert_eq!(cfg.model_override.as_deref(), Some("claude-sonnet-4-5"));
1397            assert!(cfg.keep_warm);
1398            let back = serde_json::to_value(&cfg).unwrap();
1399            assert!(back.get("memory_mode").is_none(), "not written back: {back}");
1400        }
1401        let rs: ReloadState = serde_json::from_value(serde_json::json!({
1402            "generation": 2,
1403            "written_at": "2026-10-14T18:17:03Z",
1404            "sessions": [{
1405                "id": "s1", "journal_id": "s1", "keep_warm": false, "lifecycle": "live",
1406                "config": { "memory_mode": "capture_and_recall", "cwd": "/srv/chat" }
1407            }],
1408            "expected_clients": 0
1409        }))
1410        .expect("the reload state still parses");
1411        assert_eq!(rs.sessions.len(), 1);
1412        let restored = restore_config(&rs.sessions[0].config, SessionLifecycle::Live, "s1");
1413        assert!(!restored.await_extensions);
1414    }
1415
1416    /// The restore without the hook wait, Parked or not.
1417    #[test]
1418    fn a_restore_keeps_the_sessions_memory_choice() {
1419        let saved = SessionConfig {
1420            await_extensions: true,
1421            ..Default::default()
1422        };
1423        for lifecycle in [SessionLifecycle::Live, SessionLifecycle::Parked, SessionLifecycle::Parking] {
1424            let restored = restore_config(&saved, lifecycle, "chat");
1425            assert!(!restored.await_extensions, "{lifecycle:?}");
1426        }
1427    }
1428
1429    /// `ReloadAborted` reaches a session whose command queue is full (the
1430    /// actor drains it) instead of being dropped, which would leave the
1431    /// session fenced until the backstop.
1432    #[tokio::test]
1433    async fn reload_aborted_waits_for_room_in_a_full_queue() {
1434        use crate::session::handle::echo::{meta_for, view_for};
1435        let id = crate::session::SessionId("abort-full".into());
1436        let (handle, mut ep) = SessionHandle::new(meta_for(&id), view_for());
1437        while handle.send(SessionCommand::Save).await.is_ok() {}
1438        let drain = tokio::spawn(async move {
1439            tokio::time::sleep(Duration::from_millis(100)).await;
1440            let mut got_abort = false;
1441            while let Some(a) = ep.cmd_rx.recv().await {
1442                if matches!(a.cmd, SessionCommand::ReloadAborted) {
1443                    got_abort = true;
1444                    break;
1445                }
1446            }
1447            got_abort
1448        });
1449        assert!(send_reload_aborted(&handle, Duration::from_secs(5)).await);
1450        assert!(drain.await.unwrap(), "delivered");
1451        // No room ever: gives up within its budget, reporting it.
1452        let id = crate::session::SessionId("abort-stuck".into());
1453        let (stuck, _ep) = SessionHandle::new(meta_for(&id), view_for());
1454        while stuck.send(SessionCommand::Save).await.is_ok() {}
1455        assert!(!send_reload_aborted(&stuck, Duration::from_millis(100)).await);
1456    }
1457}