diff --git a/docs/architecture/file-structure.md b/docs/architecture/file-structure.md index af485604a..4acf2f66a 100644 --- a/docs/architecture/file-structure.md +++ b/docs/architecture/file-structure.md @@ -82,7 +82,7 @@ Codync/ | CLI and background service | `host/src/main.rs`, `service/` | | API, caller permissions, event ordering | `host/src/api/`, `hub.rs` | | SQLite transcript, bots, lanes and sessions | `host/src/store/` (`entries.rs`, `bots.rs`, `model.rs`, `devices.rs`, `schema.rs`) | -| Agent process, ACP, queue and session lifecycle | `host/src/agent/bot/` (`queue.rs`, `session.rs`, `turn.rs`, `updates.rs`), `acp.rs` | +| Agent process, ACP, queue and session lifecycle | `host/src/agent/bot/` (`queue.rs`, `session.rs`, `session_setup.rs`, `turn.rs`, `updates.rs`), `acp.rs`, `process.rs`, `process_windows.rs` | | Group room turns / bot-to-bot requests | `host/src/chat/group.rs` / `team.rs` | | Prompt snapshots and memory keeper | `host/src/chat/context.rs`, `memory/` (`engram/`, `manage.rs`, `read.rs`, `extract.rs`, `keeper.rs`, `search.rs`) | | Identity, encryption and direct channel | `host/src/remote/identity.rs`, `crypto.rs`, `channel.rs` | diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index a322235dc..4b199890d 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -42,7 +42,7 @@ Cloud account metadata includes names, public keys, ownership and access state. `api/` and `remote/channel.rs` share dispatch and caller checks. `hub.rs` coordinates actors and event publication; `store/` persists mutations and the global revision. Clients catch up from `rev`, then follow events. Duplicate sends are identified by `clientNonce`. -`agent/bot/` serializes each bot's turns and owns its ACP process. A main chat and its reply threads can have distinct sessions. Group turns are queued on member bots and use their main sessions; the group itself has no harness. See [groups and threads](../features/groups-and-threads.md), [collaboration](../features/bot-collaboration.md) and [context/memory](../features/context-and-memory.md). +`agent/bot/` serializes each bot's turns and owns its ACP process. Process-tree ownership, graceful shutdown and first-prompt persistence are described in [agent lifecycle](../reference/agent-lifecycle.md). A main chat and its reply threads can have distinct sessions. Group turns are queued on member bots and use their main sessions; the group itself has no harness. See [groups and threads](../features/groups-and-threads.md), [collaboration](../features/bot-collaboration.md) and [context/memory](../features/context-and-memory.md). ## Client state and routing diff --git a/docs/reference/agent-lifecycle.md b/docs/reference/agent-lifecycle.md new file mode 100644 index 000000000..f1f1f9ad2 --- /dev/null +++ b/docs/reference/agent-lifecycle.md @@ -0,0 +1,35 @@ +# Agent process ownership + +The host owns each ACP adapter and its descendants for the adapter's lifetime. +On Unix, the adapter starts in its own process group. A monitor observes leader +exit with `waitid(WNOWAIT)` and kills the group before reaping the leader, so +cleanup cannot target a reused process ID. Dropping the owner also terminates +the group. Descendants that deliberately create a separate process group remain +outside this Unix group; idle retirement checks for those background tools. +On Windows, the shell starts suspended, joins a job with +`KILL_ON_JOB_CLOSE`, then resumes. Closing the job releases its descendants. + +Intentional shutdown first closes ACP stdin and retains process ownership during +the existing ten-second grace period, allowing CLI login refreshes to finish. +After that deadline the host terminates remaining descendants and reaps the +leader. Unexpected leader exit and actor unwinding release the owned tree. + +Abandoning an ACP answer removes its pending request. An interrupted session +allocation is marked uncertain until its identity is confirmed. The next turn +restarts a resumable adapter before allocating again; an adapter without +`loadSession` keeps confirmed conversations and requires an explicit reset +before another uncertain allocation. Fast mode and idle session closure use a +separate request path and never mark an allocation uncertain. + +A session snapshot persists the first-prompt profile and thread introduction. +The host records a new or forked identity before model setup, so setup failure +can retry the same conversation. Initial instructions clear only after the +prompt is written; they survive another lane running or a host restart. + +The existing idle policy remains: sessions unused for thirty minutes may close +when supported, and an entirely idle resumable adapter may stop. A queue, +approval, active turn or detached background descendant holds it open. Agents +without `loadSession` remain live. There is no per-process session cap. +A failed saved-session load retains the existing fresh-session fallback with a +divider notice, for both main chats and threads. Fork-only adapters retain their +existing thread-fork support. diff --git a/host/Cargo.lock b/host/Cargo.lock index ba09714a8..3a179d7bc 100644 --- a/host/Cargo.lock +++ b/host/Cargo.lock @@ -627,6 +627,7 @@ dependencies = [ "tracing-subscriber", "unicode-width", "uuid", + "windows-sys 0.61.2", "x25519-dalek", "zeroize", ] diff --git a/host/Cargo.toml b/host/Cargo.toml index 04e2fb3fa..bcf438ade 100644 --- a/host/Cargo.toml +++ b/host/Cargo.toml @@ -59,6 +59,7 @@ keyring = { version = "3.6.3", features = ["sync-secret-service", "crypto-rust", [target.'cfg(windows)'.dependencies] keyring = { version = "3.6.3", features = ["windows-native"] } +windows-sys = { version = "0.61.2", features = ["Win32_Foundation", "Win32_System_JobObjects", "Win32_System_Threading", "Win32_System_Diagnostics_ToolHelp", "Win32_Security"] } # rust-skills lint-* rules: all default groups plus pedantic, with the # library-oriented pedantic lints turned off for this binary crate. diff --git a/host/src/agent/acp.rs b/host/src/agent/acp.rs index f7303ac11..e705058bc 100644 --- a/host/src/agent/acp.rs +++ b/host/src/agent/acp.rs @@ -2,6 +2,7 @@ //! process's stdio. Updates are kept as `serde_json::Value` on purpose so a //! newer adapter adding a variant never breaks parsing. +use super::process::Process; use crate::LockExt; use anyhow::{Context, Result}; use serde_json::{Value, json}; @@ -11,9 +12,8 @@ use std::sync::Arc; use std::sync::atomic::{AtomicI64, Ordering}; use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::process::{Child, ChildStdin, Command}; +use tokio::process::ChildStdin; use tokio::sync::{mpsc, oneshot}; -use tokio::time::Instant; /// stderr lines kept to explain a crash. const STDERR_TAIL_LINES: usize = 12; @@ -93,12 +93,15 @@ impl std::error::Error for Failure {} /// A request already written to the agent, waiting for its response. pub struct Answer { - rx: oneshot::Receiver>, + id: i64, + pending: Pending, + rx: Option>>, } impl Answer { - pub async fn response(self) -> Result { - match self.rx.await { + pub async fn response(mut self) -> Result { + let receiver = self.rx.take().expect("a response is consumed once"); + match receiver.await { Ok(Ok(v)) => Ok(v), Ok(Err(e)) => { let msg = e["message"].as_str().map_or_else(|| e.to_string(), str::to_owned); @@ -114,13 +117,18 @@ impl Answer { } } +impl Drop for Answer { + fn drop(&mut self) { + self.pending.locked().remove(&self.id); + } +} + pub struct Acp { /// `None` once stopped: the closed pipe tells the agent to exit. stdin: tokio::sync::Mutex>, next_id: AtomicI64, pending: Pending, - /// `None` once stopped (handed to [`reap`]). - child: tokio::sync::Mutex>, + child: Arc>, /// The process (and, on Unix, process group) leader. pid: Option, } @@ -144,10 +152,16 @@ impl Acp { // Its own group, so `kill` can stop everything it started. #[cfg(unix)] process.process_group(0); + #[cfg(windows)] + process.creation_flags(windows_sys::Win32::System::Threading::CREATE_SUSPENDED); let mut child = process.spawn().with_context(|| format!("starting `{command}`"))?; let stdin = child.stdin.take().expect("stdin is piped"); let stdout = child.stdout.take().expect("stdout is piped"); let stderr = child.stderr.take().expect("stderr is piped"); + let pid = child.id(); + let child = Process::new(child)?; + let child = Arc::new(tokio::sync::Mutex::new(child)); + Process::monitor(&child); // Deliberately unbounded: `session/load` replays a whole history as notifications // *before* its response, while the bot is still awaiting that response. A bounded // queue would stall this reader, and with it the response — a deadlock. @@ -207,12 +221,11 @@ impl Acp { }); } - let pid = child.id(); let acp = Arc::new(Self { stdin: tokio::sync::Mutex::new(Some(stdin)), next_id: AtomicI64::new(1), pending, - child: tokio::sync::Mutex::new(Some(child)), + child, pid, }); Ok((acp, rx)) @@ -233,7 +246,8 @@ impl Acp { } pub async fn request(&self, method: &str, params: Value) -> Result { - self.send(method, params).await?.response().await + let answer = self.send(method, params).await?; + answer.response().await } /// Writes a request and returns its pending answer: anything written after this @@ -242,8 +256,9 @@ impl Acp { let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (tx, rx) = oneshot::channel(); self.pending.locked().insert(id, tx); + let answer = Answer { id, pending: self.pending.clone(), rx: Some(rx) }; self.write(json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params})).await?; - Ok(Answer { rx }) + Ok(answer) } pub async fn notify(&self, method: &str, params: Value) -> Result<()> { @@ -262,11 +277,18 @@ impl Acp { /// lets it exit by itself first: Claude Code killed during an OAuth refresh loses the new /// refresh token, and the user's login with it. pub async fn kill(&self) { - self.stdin.lock().await.take(); - if let Some(child) = self.child.lock().await.take() { + let mut child = self.child.lock().await; + if child.begin_stop() { + let mut stdin = self.stdin.lock().await; + stdin.take(); + let owned = self.child.clone(); let mut stopping = STOPPING.locked(); stopping.retain(|task| !task.is_finished()); - stopping.push(tokio::spawn(reap(child))); + stopping.push(tokio::spawn(async move { + tokio::time::sleep(EXIT_GRACE).await; + let mut process = owned.lock().await; + process.kill().await; + })); } } } @@ -277,52 +299,6 @@ pub async fn stopped() { futures::future::join_all(tasks).await; } -/// Gives the agent [`EXIT_GRACE`] to exit with everything it started (ACP adapters spawn -/// CLIs, MCP servers and tools), then kills its process group (Windows: process tree). -async fn reap(mut child: Child) { - let Some(pid) = child.id() else { return }; - let deadline = Instant::now() + EXIT_GRACE; - let exited = tokio::time::timeout_at(deadline, child.wait()).await.is_ok(); - if !exited || outlived(pid, deadline).await { - let mut tree = if cfg!(windows) { - let mut c = Command::new("taskkill"); - c.args(["/T", "/F", "/PID", &pid.to_string()]); - c - } else { - let mut c = Command::new("/bin/kill"); - c.args(["-KILL", "--", &format!("-{pid}")]); - c - }; - let _ = tree.stdout(Stdio::null()).stderr(Stdio::null()).status().await; - } - // Already exited is fine; anything else is worth a log line. - if let Err(error) = child.kill().await - && error.kind() != std::io::ErrorKind::InvalidInput - { - tracing::debug!(%error, "couldn't kill agent process"); - } -} - -/// Whether part of the exited agent's process group is still running at `deadline`. -#[cfg(unix)] -async fn outlived(pid: u32, deadline: Instant) -> bool { - let Ok(group) = libc::pid_t::try_from(pid) else { return true }; - // SAFETY: signal 0 sends nothing; it only checks that the group still has a process. - let running = || unsafe { libc::killpg(group, 0) } == 0; - while running() && Instant::now() < deadline { - tokio::time::sleep(Duration::from_millis(100)).await; - } - running() -} - -/// Windows has no process groups: what the agent started got the same closed pipes, and its -/// tree can't be found once it's gone. -#[cfg(windows)] -#[allow(clippy::unused_async)] // the Unix check's signature -async fn outlived(_pid: u32, _deadline: Instant) -> bool { - false -} - /// Plain-text rendering of an ACP content block list / tool-call content list. pub fn content_text(v: &Value) -> String { match v { @@ -355,6 +331,63 @@ pub fn truncate(s: &str, max: usize) -> String { mod tests { use super::*; + async fn descendant_agent() -> (Arc, mpsc::UnboundedReceiver) { + let script = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/process_agent.py"); + let python = if cfg!(windows) { "python" } else { "python3" }; + let command = format!("{python} -u {}", crate::shell::quote(&script.display().to_string())); + let (acp, mut rx) = Acp::spawn(&command, env!("CARGO_MANIFEST_DIR"), &[]).unwrap(); + let ready = tokio::time::timeout(Duration::from_secs(15), rx.recv()).await; + let ready = ready.unwrap(); + assert!(matches!(ready, Some(Incoming::Notification { method, .. }) if method == "ready")); + (acp, rx) + } + + async fn assert_descendant_pipes_close(mut rx: mpsc::UnboundedReceiver) { + let closed = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await; + let closed = closed.unwrap(); + assert!(matches!(closed, Some(Incoming::Closed { .. }))); + } + + #[tokio::test] + async fn leader_exit_releases_descendants_without_waiting_for_stdout_eof() { + let (acp, rx) = descendant_agent().await; + let notified = acp.notify("exit", json!({})).await; + notified.unwrap(); + assert_descendant_pipes_close(rx).await; + } + + #[tokio::test] + async fn dropping_an_agent_releases_its_process_tree() { + let (acp, rx) = descendant_agent().await; + drop(acp); + assert_descendant_pipes_close(rx).await; + } + + #[tokio::test] + async fn actor_unwinding_releases_its_process_tree() { + let (acp, rx) = descendant_agent().await; + let task = tokio::spawn(async move { + let _owned = acp; + panic!("simulated actor panic"); + }); + let panicked = task.await; + assert!(panicked.unwrap_err().is_panic()); + assert_descendant_pipes_close(rx).await; + } + + #[tokio::test] + async fn timed_out_requests_do_not_accumulate_pending_answers() { + let (acp, rx) = descendant_agent().await; + for _ in 0..10 { + let response = tokio::time::timeout(Duration::from_millis(10), acp.request("never", json!({}))).await; + assert!(response.is_err()); + assert_eq!(acp.pending.locked().len(), 0); + } + acp.kill().await; + stopped().await; + assert_descendant_pipes_close(rx).await; + } + #[test] fn content_text_flattens_blocks() { let v = json!([ diff --git a/host/src/agent/bot/fast_mode.rs b/host/src/agent/bot/fast_mode.rs index 927f88d14..e72e2c8c8 100644 --- a/host/src/agent/bot/fast_mode.rs +++ b/host/src/agent/bot/fast_mode.rs @@ -55,7 +55,8 @@ impl Actor { let Some(option) = self.conn.as_ref().and_then(|c| c.fast_options.get(sid)) else { return }; let (Some(config_id), Some(value)) = (option["id"].as_str(), fast_mode_change(option, on)) else { return }; let params = json!({"sessionId": sid, "configId": config_id, "value": value}); - match self.session_request("session/set_config_option", params).await { + let switched = self.request_with_timeout("session/set_config_option", params).await; + match switched { Ok(answer) => self.note_options(sid, &answer, Listing::Switch), Err(error) => { tracing::warn!(bot = %self.cfg.id, error = format!("{error:#}"), "couldn't set Fast mode"); diff --git a/host/src/agent/bot/lifecycle.rs b/host/src/agent/bot/lifecycle.rs index bf4bfd6aa..b3c7c1310 100644 --- a/host/src/agent/bot/lifecycle.rs +++ b/host/src/agent/bot/lifecycle.rs @@ -83,13 +83,14 @@ impl Actor { } async fn close_session(&mut self, sid: &str) { - let closed = self.session_request("session/close", json!({ "sessionId": sid })).await; + let closed = self.request_with_timeout("session/close", json!({ "sessionId": sid })).await; let Some(conn) = self.conn.as_mut() else { return }; match closed { Ok(_) => { tracing::info!(bot = %self.cfg.id, session = sid, "closed an idle session"); conn.loaded.remove(sid); conn.fast_options.remove(sid); + conn.pending_models.remove(sid); } Err(error) => { tracing::warn!(bot = %self.cfg.id, session = sid, error = format!("{error:#}"), "couldn't close an idle session"); diff --git a/host/src/agent/bot/mod.rs b/host/src/agent/bot/mod.rs index 02628c3ca..7266468e9 100644 --- a/host/src/agent/bot/mod.rs +++ b/host/src/agent/bot/mod.rs @@ -20,6 +20,7 @@ mod files; mod lifecycle; mod queue; mod session; +mod session_setup; mod turn; mod updates; @@ -126,13 +127,11 @@ pub fn spawn(hub: Arc, cfg: BotConfig) -> BotHandle { thread_sessions: HashMap::new(), lane, turn_session: None, - thread_intro: None, active_group: None, active_routine: None, routine_deadline: None, routine_timed_out: false, routine_completion: None, - session_fresh: false, applied_system: HashMap::new(), keeper, turn_text: None, @@ -187,6 +186,9 @@ pub(crate) struct Conn { /// Sessions already created or loaded in this process, and when each was last used. loaded: HashMap, started: Instant, + allocation_uncertain: bool, + /// Allocated sessions whose model setup has not yet completed. + pending_models: HashMap, /// Each live session's Fast mode option, as the agent last listed it. fast_options: HashMap, } @@ -216,15 +218,11 @@ struct Actor { lane: Lane, /// The session the current turn prompts. turn_session: Option, - /// Told on the first message of a new thread session: what the thread is about. - thread_intro: Option, active_group: Option, active_routine: Option, routine_deadline: Option, routine_timed_out: bool, routine_completion: Option, - /// True until the first prompt of a new ACP session has been sent. - session_fresh: bool, /// Session -> the instructions this process last handed Claude as its system prompt. applied_system: HashMap, keeper: mpsc::UnboundedSender, diff --git a/host/src/agent/bot/session.rs b/host/src/agent/bot/session.rs index 007a8a7c4..4169113b6 100644 --- a/host/src/agent/bot/session.rs +++ b/host/src/agent/bot/session.rs @@ -1,11 +1,9 @@ //! The agent process and its ACP sessions: launch, handshake, new/load/fork, thread sessions. -use super::fast_mode::Listing; use super::{Actor, Conn, NoticeStyle, SessionCaps}; -use crate::agent::acp::{self, Acp, Incoming}; -use crate::chat::context; +use crate::agent::acp::{self, Acp}; use crate::store::{BotConfig, Entry, EntryKind, Lane, lane_key}; -use anyhow::{Result, anyhow, bail}; +use anyhow::{Result, anyhow}; use serde_json::{Value, json}; use std::collections::HashMap; use std::time::{Duration, Instant}; @@ -131,187 +129,6 @@ impl Actor { Ok(Value::Array(servers)) } - /// Makes sure the agent runs and `slot`'s session (main, or a thread's) is live; - /// sets `turn_session`. A thread without one forks the main session when it can. - pub(super) async fn ensure_session(&mut self, slot: Option<&str>) -> Result<()> { - // Memory is best-effort: a bot without it still answers. - if let Err(error) = crate::chat::memory::prepare(&self.cfg.id).await { - self.memory_unavailable(&error); - } - if !self.cfg.connectors.is_empty() - || self.hub.store.kv_read(&format!("agent-env:{}", self.cfg.backend))?.is_some() - { - crate::market::vault::unlock(self.hub.clone()).await?; - } - // An agent that can't resume keeps its session; its next new session gets the new tools. - if std::mem::take(&mut self.tools_changed) - && self.conn.as_ref().is_some_and(|c| c.sessions.load) - && let Some(c) = self.conn.take() - { - c.acp.kill().await; - } - if self.conn.is_none() { - let (hub, id) = (self.hub.clone(), self.cfg.id.clone()); - let candidates = launch_commands(&self.cfg, move |msg: &str| { - let msg = msg.to_owned(); - hub.set_runtime(&id, |r| r.activity = msg); - }) - .await?; - self.hub.set_runtime(&self.id(), |r| r.activity = "Starting agent…".into()); - // Keys saved from an "environment variable" sign-in. - let env = crate::agent::auth::env(&self.hub.store, &self.cfg.backend)?; - let mut last_err = None; - for (i, command) in candidates.iter().enumerate() { - let fallback_left = i + 1 < candidates.len(); - // A local CLI too old for ACP just hangs; don't make the user wait long before the fallback. - let budget = Duration::from_secs(if fallback_left { 20 } else { 180 }); - match start_agent(command, &self.cfg.cwd, &env, budget).await { - Ok(conn) => { - self.conn = Some(conn); - break; - } - Err(e) => { - tracing::info!(command, error = format!("{e:#}"), "agent failed to start"); - last_err = Some(e); - } - } - } - if self.conn.is_none() { - return Err(last_err.unwrap_or_else(|| anyhow!("agent failed to start"))); - } - } - let servers = self.mcp_servers()?; - let claude = self.conn.as_ref().is_some_and(|c| c.claude); - let mut forked = false; - if let Some(root) = slot - && self.session_for(slot).is_none() - && let Some(main) = self.session_id.clone() - && self.conn.as_ref().is_some_and(|c| c.sessions.fork) - { - self.hub.set_runtime(&self.cfg.id, |r| r.activity = "Opening the thread…".into()); - let res = self - .session_request("session/fork", json!({"sessionId": main, "cwd": self.cfg.cwd, "mcpServers": servers})) - .await - .ok(); - if let Some(res) = res - && let Some(sid) = res["sessionId"].as_str().map(str::to_owned) - && let Some(conn) = self.conn.as_mut() - { - conn.loaded.insert(sid.clone(), Instant::now()); - self.set_session(slot, Some(&sid)); - self.note_options(&sid, &res, Listing::Session); - forked = true; - } else { - tracing::warn!(bot = %self.cfg.id, "session/fork failed; the thread starts fresh"); - } - if forked { - self.thread_intro = Some(self.thread_intro(root, true)); - } - } - // Claude gets the frozen instructions as its system prompt. When a compaction - // re-rendered them, loading the session again swaps the prompt in (the adapter - // rebuilds its query and resumes the same conversation). - let current = self.session_for(slot); - let snapshot = match (¤t, claude) { - (Some(sid), true) => Some(self.snapshot(sid).await?), - _ => None, - }; - let Some(conn) = self.conn.as_ref() else { bail!("agent not running") }; - if let Some(sid) = current { - let prompt_current = - snapshot.as_ref().is_none_or(|s| self.applied_system.get(&sid).is_some_and(|t| *t == s.system)); - if conn.loaded.contains_key(&sid) && prompt_current { - self.turn_session = Some(sid); - return Ok(()); - } - if !conn.sessions.load { - self.notice( - "This agent can't resume sessions after a restart — starting a fresh one.", - NoticeStyle::Divider, - ); - self.set_session(slot, None); - return Box::pin(self.ensure_session(slot)).await; - } - self.hub.set_runtime(&self.cfg.id, |r| r.activity = "Resuming session…".into()); - let mut params = json!({"sessionId": sid, "cwd": self.cfg.cwd, "mcpServers": servers}); - if let Some(s) = &snapshot { - params["_meta"] = system_meta(&s.system); - } - let res = self.session_request("session/load", params).await; - let Some(conn) = self.conn.as_mut() else { return res.map(drop) }; - // The agent replays history as session/update before answering; drop it. - while let Ok(inc) = conn.rx.try_recv() { - if let Incoming::Request { id, .. } = inc { - let _ = conn.acp.respond_error(id, -32601, "not supported").await; - } - } - match res { - Ok(res) => { - // A resumed session can come back on the agent's last-used model (Cursor - // keeps one for all its sessions), so the bot's own is set again. - let res = if res.get("configOptions").is_some() || res.get("models").is_some() { - select_model(&conn.acp, &res, &sid, &self.cfg).await? - } else { - res - }; - conn.loaded.insert(sid.clone(), Instant::now()); - self.note_options(&sid, &res, Listing::Session); - if let Some(s) = snapshot { - // Its system prompt now names the bot as rendered. - context::mark_announced(&self.hub.store, &self.cfg.id, &s, s.system_identity.clone())?; - self.applied_system.insert(sid.clone(), s.system); - } - self.turn_session = Some(sid); - return Ok(()); - } - Err(e) => { - tracing::warn!(error = format!("{e:#}"), "session/load failed; starting fresh"); - self.thread_intro = None; - self.notice("Couldn't resume the previous session — starting a fresh one.", NoticeStyle::Divider); - } - } - } - let system = { - let (hub, cfg) = (self.hub.clone(), self.cfg.clone()); - tokio::task::spawn_blocking(move || context::render(&hub.store, &cfg)).await? - }; - let mut params = json!({"cwd": self.cfg.cwd, "mcpServers": servers}); - if claude { - params["_meta"] = system_meta(&system); - } - let res = self.session_request("session/new", params).await?; - let Some(conn) = self.conn.as_mut() else { bail!("agent not running") }; - let sid = res["sessionId"].as_str().ok_or_else(|| anyhow!("agent returned no sessionId"))?.to_owned(); - conn.loaded.insert(sid.clone(), Instant::now()); - let options = select_model(&conn.acp, &res, &sid, &self.cfg).await?; - self.note_options(&sid, &options, Listing::Session); - self.set_session(slot, Some(&sid)); - self.session_fresh = true; - self.turn_session = Some(sid.clone()); - if let Some(root) = slot { - self.thread_intro = Some(self.thread_intro(root, false)); - } - context::adopt(&self.hub.store, &self.cfg, &sid, system.clone())?; - if claude { - self.applied_system.insert(sid, system); - } - Ok(()) - } - - /// A session request the agent doesn't answer in time stops the agent: a hung one - /// would otherwise hold the bot, and its Stop, forever. - pub(super) async fn session_request(&mut self, method: &str, params: Value) -> Result { - let acp = self.conn.as_ref().ok_or_else(|| anyhow!("agent not running"))?.acp.clone(); - if let Ok(result) = tokio::time::timeout(SESSION_TIMEOUT, acp.request(method, params)).await { - return result; - } - if let Some(c) = self.conn.take() { - c.acp.kill().await; - } - let message = format!("the agent didn't answer {method} within {}s", SESSION_TIMEOUT.as_secs()); - Err(acp::Failure::error(acp::FailureKind::Timeout, message)) - } - /// Memory failed to start: said once per actor, then the bot works without it. pub(super) fn memory_unavailable(&mut self, error: &anyhow::Error) { tracing::warn!(bot = %self.cfg.id, error = format!("{error:#}"), "memory unavailable"); @@ -325,15 +142,15 @@ impl Actor { /// Drops the main ACP session so the next turn starts a fresh one. pub(super) fn forget_session(&mut self) { + if self.conn.as_ref().is_some_and(|conn| conn.allocation_uncertain) { + self.conn = None; + } self.set_session(None, None); } } -/// How long `session/new`, `session/load` and `session/fork` may take. -const SESSION_TIMEOUT: Duration = Duration::from_secs(180); - /// Claude's `_meta`: the bot's instructions appended to Claude Code's own system prompt. -fn system_meta(system: &str) -> Value { +pub(super) fn system_meta(system: &str) -> Value { json!({"systemPrompt": {"append": system}}) } @@ -442,6 +259,8 @@ pub(crate) async fn start_agent(command: &str, cwd: &str, env: &[(String, String claude, loaded: HashMap::new(), started: Instant::now(), + allocation_uncertain: false, + pending_models: HashMap::new(), fast_options: HashMap::new(), }) } diff --git a/host/src/agent/bot/session_setup.rs b/host/src/agent/bot/session_setup.rs new file mode 100644 index 000000000..ab9beb5b9 --- /dev/null +++ b/host/src/agent/bot/session_setup.rs @@ -0,0 +1,227 @@ +//! Session allocation and resumption, retaining first-prompt instructions across setup failures. + +use super::fast_mode::Listing; +use super::session::system_meta; +use super::{Actor, NoticeStyle, launch_commands, select_model, start_agent}; +use crate::agent::acp::{self, Incoming}; +use crate::chat::context; +use anyhow::{Result, anyhow, bail}; +use serde_json::{Value, json}; +use std::time::{Duration, Instant}; + +const SESSION_TIMEOUT: Duration = Duration::from_secs(180); + +impl Actor { + /// Makes sure the agent runs and `slot`'s session (main, or a thread's) is live; + /// sets `turn_session`. A thread without one forks the main session when it can. + pub(super) async fn ensure_session(&mut self, slot: Option<&str>) -> Result<()> { + // A cancelled allocation may have started tools without returning their session ID. + // Resumable agents can be replaced; nonresumable agents retain confirmed conversations. + if self.conn.as_ref().is_some_and(|conn| conn.allocation_uncertain && conn.sessions.load) + && let Some(conn) = self.conn.take() + { + conn.acp.kill().await; + } + // Memory is best-effort: a bot without it still answers. + if let Err(error) = crate::chat::memory::prepare(&self.cfg.id).await { + self.memory_unavailable(&error); + } + if !self.cfg.connectors.is_empty() + || self.hub.store.kv_read(&format!("agent-env:{}", self.cfg.backend))?.is_some() + { + crate::market::vault::unlock(self.hub.clone()).await?; + } + // An agent that can't resume keeps its session; its next new session gets the new tools. + if std::mem::take(&mut self.tools_changed) + && self.conn.as_ref().is_some_and(|c| c.sessions.load) + && let Some(c) = self.conn.take() + { + c.acp.kill().await; + } + if self.conn.is_none() { + let (hub, id) = (self.hub.clone(), self.cfg.id.clone()); + let candidates = launch_commands(&self.cfg, move |msg: &str| { + let msg = msg.to_owned(); + hub.set_runtime(&id, |r| r.activity = msg); + }) + .await?; + self.hub.set_runtime(&self.id(), |r| r.activity = "Starting agent…".into()); + // Keys saved from an "environment variable" sign-in. + let env = crate::agent::auth::env(&self.hub.store, &self.cfg.backend)?; + let mut last_err = None; + for (i, command) in candidates.iter().enumerate() { + let fallback_left = i + 1 < candidates.len(); + // A local CLI too old for ACP just hangs; don't make the user wait long before the fallback. + let budget = Duration::from_secs(if fallback_left { 20 } else { 180 }); + let started = start_agent(command, &self.cfg.cwd, &env, budget).await; + match started { + Ok(conn) => { + self.conn = Some(conn); + break; + } + Err(e) => { + tracing::info!(command, error = format!("{e:#}"), "agent failed to start"); + last_err = Some(e); + } + } + } + if self.conn.is_none() { + return Err(last_err.unwrap_or_else(|| anyhow!("agent failed to start"))); + } + } + let servers = self.mcp_servers()?; + let claude = self.conn.as_ref().is_some_and(|c| c.claude); + if slot.is_some() + && self.session_for(slot).is_none() + && let Some(main) = self.session_id.clone() + && self.conn.as_ref().is_some_and(|c| c.sessions.fork) + { + self.hub.set_runtime(&self.cfg.id, |r| r.activity = "Opening the thread…".into()); + let response = self + .session_request("session/fork", json!({"sessionId": main, "cwd": self.cfg.cwd, "mcpServers": servers})) + .await; + let res = response.ok(); + if let Some(res) = res + && let Some(sid) = + res["sessionId"].as_str().filter(|sid| !sid.is_empty() && **sid != main).map(str::to_owned) + && let Some(conn) = self.conn.as_mut() + { + conn.loaded.insert(sid.clone(), Instant::now()); + conn.allocation_uncertain = false; + conn.pending_models.insert(sid.clone(), res.clone()); + let parent = self.snapshot(&main).await?; + context::fork(&self.hub.store, &self.cfg, &parent, &sid)?; + self.set_session(slot, Some(&sid)); + self.note_options(&sid, &res, Listing::Session); + } else { + tracing::warn!(bot = %self.cfg.id, "session/fork failed; the thread starts fresh"); + } + } + // Claude gets the frozen instructions as its system prompt. When a compaction + // re-rendered them, loading the session again swaps the prompt in (the adapter + // rebuilds its query and resumes the same conversation). + let current = self.session_for(slot); + let snapshot = match (¤t, claude) { + (Some(sid), true) => Some(self.snapshot(sid).await?), + _ => None, + }; + let Some(conn) = self.conn.as_ref() else { bail!("agent not running") }; + if let Some(sid) = current { + let prompt_current = + snapshot.as_ref().is_none_or(|s| self.applied_system.get(&sid).is_some_and(|t| *t == s.system)); + if conn.loaded.contains_key(&sid) && prompt_current { + self.finish_session_model(&sid).await?; + self.turn_session = Some(sid); + return Ok(()); + } + if !conn.sessions.load { + self.notice( + "This agent can't resume sessions after a restart — starting a fresh one.", + NoticeStyle::Divider, + ); + self.set_session(slot, None); + return Box::pin(self.ensure_session(slot)).await; + } + self.hub.set_runtime(&self.cfg.id, |r| r.activity = "Resuming session…".into()); + let mut params = json!({"sessionId": sid, "cwd": self.cfg.cwd, "mcpServers": servers}); + if let Some(s) = &snapshot { + params["_meta"] = system_meta(&s.system); + } + let res = self.session_request("session/load", params).await; + let Some(conn) = self.conn.as_mut() else { return res.map(drop) }; + // The agent replays history as session/update before answering; drop it. + while let Ok(inc) = conn.rx.try_recv() { + if let Incoming::Request { id, .. } = inc { + let _ = conn.acp.respond_error(id, -32601, "not supported").await; + } + } + match res { + Ok(res) => { + conn.loaded.insert(sid.clone(), Instant::now()); + conn.allocation_uncertain = false; + if res.get("configOptions").is_some() || res.get("models").is_some() { + conn.pending_models.insert(sid.clone(), res.clone()); + } + self.note_options(&sid, &res, Listing::Session); + if let Some(s) = snapshot { + // Its system prompt now names the bot as rendered. + context::mark_announced(&self.hub.store, &self.cfg.id, &s, s.system_identity.clone())?; + self.applied_system.insert(sid.clone(), s.system); + } + self.finish_session_model(&sid).await?; + self.turn_session = Some(sid); + return Ok(()); + } + Err(e) => { + tracing::warn!(error = format!("{e:#}"), "session/load failed; starting fresh"); + self.notice("Couldn't resume the previous session — starting a fresh one.", NoticeStyle::Divider); + } + } + } + let system = { + let (hub, cfg) = (self.hub.clone(), self.cfg.clone()); + tokio::task::spawn_blocking(move || context::render(&hub.store, &cfg)).await? + }; + let mut params = json!({"cwd": self.cfg.cwd, "mcpServers": servers}); + if claude { + params["_meta"] = system_meta(&system); + } + let res = self.session_request("session/new", params).await?; + let Some(conn) = self.conn.as_mut() else { bail!("agent not running") }; + let sid = res["sessionId"].as_str().ok_or_else(|| anyhow!("agent returned no sessionId"))?.to_owned(); + conn.loaded.insert(sid.clone(), Instant::now()); + conn.allocation_uncertain = false; + conn.pending_models.insert(sid.clone(), res); + self.set_session(slot, Some(&sid)); + context::adopt_new(&self.hub.store, &self.cfg, &sid, system.clone())?; + if claude { + self.applied_system.insert(sid.clone(), system); + } + self.finish_session_model(&sid).await?; + self.turn_session = Some(sid); + Ok(()) + } + + /// Only allocations can leave resources with an unknown session identity when cancelled. + pub(super) async fn session_request(&mut self, method: &str, params: Value) -> Result { + let conn = self.conn.as_mut().ok_or_else(|| anyhow!("agent not running"))?; + if conn.allocation_uncertain { + bail!( + "The agent did not confirm its previous allocation. Existing conversations are preserved; reset the affected session before opening another." + ); + } + conn.allocation_uncertain = true; + let result = self.request_with_timeout(method, params).await; + if result.is_err() + && let Some(conn) = self.conn.as_mut() + { + conn.allocation_uncertain = false; + } + result + } + + /// Mode changes and idle closes do not allocate sessions. + pub(super) async fn request_with_timeout(&mut self, method: &str, params: Value) -> Result { + let acp = self.conn.as_ref().ok_or_else(|| anyhow!("agent not running"))?.acp.clone(); + let response = tokio::time::timeout(SESSION_TIMEOUT, acp.request(method, params)).await; + if let Ok(result) = response { + return result; + } + if let Some(conn) = self.conn.take() { + conn.acp.kill().await; + } + let message = format!("the agent didn't answer {method} within {}s", SESSION_TIMEOUT.as_secs()); + Err(acp::Failure::error(acp::FailureKind::Timeout, message)) + } + + async fn finish_session_model(&mut self, sid: &str) -> Result<()> { + let Some(conn) = self.conn.as_ref() else { bail!("agent not running") }; + let Some(response) = conn.pending_models.get(sid) else { return Ok(()) }; + let options = select_model(&conn.acp, response, sid, &self.cfg).await?; + self.note_options(sid, &options, Listing::Session); + if let Some(conn) = self.conn.as_mut() { + conn.pending_models.remove(sid); + } + Ok(()) + } +} diff --git a/host/src/agent/bot/turn.rs b/host/src/agent/bot/turn.rs index ff15eaa37..d1ad0f795 100644 --- a/host/src/agent/bot/turn.rs +++ b/host/src/agent/bot/turn.rs @@ -86,7 +86,10 @@ impl Actor { if let Some(sid) = self.turn_session.clone() { self.sync_fast_mode(&sid).await; } - if hidden && self.session_fresh { + let sid = self.turn_session.clone().ok_or_else(|| anyhow!("no session"))?; + let snapshot = self.snapshot(&sid).await?; + let fresh = matches!(snapshot.prelude, Some(context::Prelude::New | context::Prelude::UnpromptedFork)); + if hidden && fresh { // A new session must never silently repeat an interrupted routine. self.finish_routine( crate::routines::Status::Interrupted, @@ -108,7 +111,7 @@ impl Actor { } // Session-start chatter (banners, command lists) isn't part of the reply. // Some adapters (pi-acp) send it on a timer right after session/new. - if self.session_fresh { + if fresh { tokio::time::sleep(Duration::from_millis(400)).await; } if let Some(conn) = self.conn.as_mut() { @@ -119,7 +122,6 @@ impl Actor { } } let claude = self.conn.as_ref().is_some_and(|c| c.claude); - let sid = self.turn_session.clone().ok_or_else(|| anyhow!("no session"))?; self.memory_session = match memory::lifecycle::begin(&self.hub, &self.cfg.id, &sid).await { Ok(session) => Some(session), Err(error) => { @@ -133,17 +135,16 @@ impl Actor { { tracing::warn!(bot = %self.cfg.id, %error, "memory prompt capture failed"); } - let snapshot = self.snapshot(&sid).await?; let remind_files = !snapshot.system.contains("send_file"); let mut prompt = text.to_owned(); if let Some(notice) = memory::lifecycle::change_notice(&self.hub, &self.cfg.id, &sid) { prompt = format!("{notice}\n\n{prompt}"); } - if let Some(intro) = self.thread_intro.take() { + if let (Some(root), Some(prelude)) = (&slot, snapshot.prelude) { + let intro = self.thread_intro(root, prelude != context::Prelude::New); prompt = format!("{intro}\n\n{prompt}"); } - if self.session_fresh { - self.session_fresh = false; + if fresh { // Claude already has the instructions as its system prompt. if !claude { prompt = format!("\n{}\n\n\n{prompt}", snapshot.system); @@ -151,7 +152,7 @@ impl Actor { } if let Some((update, identity)) = context::profile_update(&self.hub.store, &snapshot, &self.cfg) { prompt = format!("{prompt}\n\n{update}"); - self.announce = Some((snapshot, identity)); + self.announce = Some((snapshot.clone(), identity)); } if remind_files && self.active_group.is_none() && self.active_request.is_none() && self.active_routine.is_none() { @@ -169,6 +170,11 @@ impl Actor { } // Written before this returns, so a Stop right after is sent after the prompt. let answer = acp.send("session/prompt", params).await?; + if snapshot.prelude.is_some() + && let Err(error) = context::mark_prompted(&self.hub.store, &self.cfg.id, &snapshot) + { + tracing::warn!(bot = %self.cfg.id, %error, "couldn't acknowledge initial session instructions"); + } if let Err(error) = memory::lifecycle::mark_announced(&self.hub, &self.cfg.id, &sid, &self.turn_memory_revision) { tracing::warn!(%error, "could not acknowledge memory update notice"); diff --git a/host/src/agent/mod.rs b/host/src/agent/mod.rs index 7c6bb7693..e9e0faf43 100644 --- a/host/src/agent/mod.rs +++ b/host/src/agent/mod.rs @@ -10,3 +10,7 @@ pub mod node; pub mod registry; pub mod term; pub mod workspace; + +mod process; +#[cfg(windows)] +mod process_windows; diff --git a/host/src/agent/process.rs b/host/src/agent/process.rs new file mode 100644 index 000000000..20f11e6f9 --- /dev/null +++ b/host/src/agent/process.rs @@ -0,0 +1,196 @@ +//! Own the agent's descendants even after its leader exits or the actor unwinds. + +use anyhow::{Context, Result}; +use std::sync::Arc; +use std::time::Duration; +use tokio::process::Child; +use tokio::sync::Mutex; + +const REAP_TIMEOUT: Duration = Duration::from_secs(5); + +pub(super) struct Process { + // Drop the tree before Child can reap the leader and make its PID reusable. + tree: Tree, + child: Child, + stopping: bool, +} + +impl Process { + pub(super) fn new(child: Child) -> Result { + let tree = Tree::new(&child)?; + let owned = Self { tree, child, stopping: false }; + #[cfg(windows)] + super::process_windows::resume(&owned.child)?; + Ok(owned) + } + + pub(super) fn monitor(child: &Arc>) { + let weak = Arc::downgrade(child); + tokio::spawn(async move { + loop { + let Some(child) = weak.upgrade() else { break }; + let mut process = child.lock().await; + if process.stopping { + break; + } + match process.exited() { + Ok(true) => { + process.kill().await; + break; + } + Ok(false) => {} + Err(error) => { + tracing::warn!(%error, "couldn't observe agent process exit"); + process.kill().await; + break; + } + } + drop(process); + drop(child); + tokio::time::sleep(Duration::from_millis(100)).await; + } + }); + } + + #[cfg(unix)] + fn exited(&self) -> std::io::Result { + let Some(pid) = self.child.id() else { return Ok(true) }; + // SAFETY: zero is a valid siginfo value; waitid writes only into this object. + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // Keep the leader unreaped until its group is killed, preventing PID reuse. + // SAFETY: pid identifies our owned child and the output pointer is writable. + let result = + unsafe { libc::waitid(libc::P_PID, pid, &raw mut info, libc::WEXITED | libc::WNOHANG | libc::WNOWAIT) }; + if result == -1 { + let error = std::io::Error::last_os_error(); + if error.kind() == std::io::ErrorKind::Interrupted { + return Ok(false); + } + return Err(error); + } + #[cfg(target_os = "macos")] + let exited = info.si_pid != 0; + #[cfg(not(target_os = "macos"))] + // SAFETY: waitid initialized the child-status variant of siginfo. + let exited = unsafe { info.si_pid() } != 0; + Ok(exited) + } + + #[cfg(windows)] + fn exited(&self) -> std::io::Result { + use windows_sys::Win32::Foundation::{WAIT_FAILED, WAIT_OBJECT_0}; + let Some(handle) = self.child.raw_handle() else { return Ok(true) }; + // SAFETY: the child owns this valid process handle; a zero timeout cannot block. + let status = unsafe { windows_sys::Win32::System::Threading::WaitForSingleObject(handle, 0) }; + if status == WAIT_FAILED { + return Err(std::io::Error::last_os_error()); + } + Ok(status == WAIT_OBJECT_0) + } + + /// The graceful-shutdown task keeps ownership until its deadline. + pub(super) fn begin_stop(&mut self) -> bool { + !std::mem::replace(&mut self.stopping, true) + } + + pub(super) async fn kill(&mut self) { + if let Err(error) = self.tree.terminate() { + tracing::warn!(%error, "couldn't terminate agent descendants"); + } + let result = tokio::time::timeout(REAP_TIMEOUT, self.child.kill()).await; + match result { + Ok(Ok(())) => {} + Ok(Err(error)) if error.kind() == std::io::ErrorKind::InvalidInput => {} + Ok(Err(error)) => tracing::warn!(%error, "couldn't reap agent process"), + Err(error) => tracing::warn!(%error, "agent process did not exit before the cleanup deadline"), + } + } +} + +#[cfg(unix)] +struct Tree { + group: Option, +} + +#[cfg(unix)] +impl Tree { + fn new(child: &Child) -> Result { + let pid = child.id().context("agent exited before process ownership was established")?; + Ok(Self { group: Some(libc::pid_t::try_from(pid)?) }) + } + + fn terminate(&mut self) -> std::io::Result<()> { + let Some(group) = self.group.take() else { return Ok(()) }; + // SAFETY: spawn put this still-owned, unreaped child in its own process group. + // A negative PID signals that group. No pointer or shared memory is involved. + if unsafe { libc::kill(-group, libc::SIGKILL) } == -1 { + let error = std::io::Error::last_os_error(); + if error.raw_os_error() != Some(libc::ESRCH) { + return Err(error); + } + } + Ok(()) + } +} + +#[cfg(unix)] +impl Drop for Tree { + fn drop(&mut self) { + if let Err(error) = self.terminate() { + tracing::warn!(%error, "couldn't release agent process group"); + } + } +} + +#[cfg(windows)] +struct Tree { + handle: std::os::windows::io::OwnedHandle, +} + +#[cfg(windows)] +impl Tree { + fn new(child: &Child) -> Result { + use std::os::windows::io::{AsRawHandle, FromRawHandle}; + use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectExtendedLimitInformation, SetInformationJobObject, + }; + // SAFETY: null uses default security and an unnamed, newly owned job. + let raw = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) }; + if raw.is_null() { + return Err(std::io::Error::last_os_error()).context("creating agent process job"); + } + // SAFETY: CreateJobObjectW transferred a unique valid handle to this owner. + let handle = unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(raw) }; + let mut limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default(); + limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + let size = u32::try_from(size_of_val(&limits))?; + // SAFETY: limits is initialized and its exact size is supplied for this class. + if unsafe { + SetInformationJobObject( + handle.as_raw_handle(), + JobObjectExtendedLimitInformation, + std::ptr::from_ref(&limits).cast(), + size, + ) + } == 0 + { + return Err(std::io::Error::last_os_error()).context("configuring agent process job"); + } + let process = child.raw_handle().context("agent exited before process ownership was established")?; + // SAFETY: the child and job handles remain owned and valid during assignment. + if unsafe { AssignProcessToJobObject(handle.as_raw_handle(), process) } == 0 { + return Err(std::io::Error::last_os_error()).context("owning agent process descendants"); + } + Ok(Self { handle }) + } + + fn terminate(&mut self) -> std::io::Result<()> { + use std::os::windows::io::AsRawHandle; + // SAFETY: this object owns the job handle; termination does not close it. + if unsafe { windows_sys::Win32::System::JobObjects::TerminateJobObject(self.handle.as_raw_handle(), 1) } == 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + } +} diff --git a/host/src/agent/process_windows.rs b/host/src/agent/process_windows.rs new file mode 100644 index 000000000..c9ce4af88 --- /dev/null +++ b/host/src/agent/process_windows.rs @@ -0,0 +1,49 @@ +//! Resume the suspended ACP shell only after its job owns future descendants. + +use anyhow::{Context, Result, bail}; +use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; +use tokio::process::Child; +use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE; +use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next, +}; +use windows_sys::Win32::System::Threading::{OpenThread, ResumeThread, THREAD_SUSPEND_RESUME}; + +pub(super) fn resume(child: &Child) -> Result<()> { + let pid = child.id().context("agent exited before it could be resumed")?; + // SAFETY: this creates an owned thread snapshot; no pointers are supplied. + let raw = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) }; + if raw == INVALID_HANDLE_VALUE { + return Err(std::io::Error::last_os_error()).context("finding suspended agent thread"); + } + // SAFETY: a successful snapshot transfers a unique valid handle to us. + let snapshot = unsafe { OwnedHandle::from_raw_handle(raw) }; + let mut entry = THREADENTRY32::default(); + entry.dwSize = u32::try_from(size_of_val(&entry))?; + // SAFETY: the snapshot is valid and entry has the required initialized size. + let mut present = unsafe { Thread32First(snapshot.as_raw_handle(), &raw mut entry) }; + while present != 0 { + if entry.th32OwnerProcessID == pid { + // The owned shell was created suspended and cannot have started more threads. + // SAFETY: this opens that thread with only permission to resume it. + let raw = unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID) }; + if raw.is_null() { + return Err(std::io::Error::last_os_error()).context("opening suspended agent thread"); + } + // SAFETY: OpenThread transferred a unique valid handle to us. + let thread = unsafe { OwnedHandle::from_raw_handle(raw) }; + // SAFETY: we own a valid handle with THREAD_SUSPEND_RESUME rights. + let count = unsafe { ResumeThread(thread.as_raw_handle()) }; + if count == u32::MAX { + return Err(std::io::Error::last_os_error()).context("resuming agent thread"); + } + if count != 1 { + bail!("agent thread had an unexpected suspension count"); + } + return Ok(()); + } + // SAFETY: the snapshot and writable entry remain valid during iteration. + present = unsafe { Thread32Next(snapshot.as_raw_handle(), &raw mut entry) }; + } + bail!("couldn't find the suspended agent's initial thread") +} diff --git a/host/src/chat/context.rs b/host/src/chat/context.rs index 829303e90..aa0657c51 100644 --- a/host/src/chat/context.rs +++ b/host/src/chat/context.rs @@ -42,6 +42,17 @@ pub struct Snapshot { pub system_identity: Identity, /// The identity the agent last heard of (`system` or a later update block). pub announced: Identity, + /// Initial instructions still owed to the first prompt, including after restart. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prelude: Option, +} + +#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Prelude { + New, + Fork, + UnpromptedFork, } // A bot talks in several lanes (its chat, threads, groups), each its own session: snapshots @@ -105,6 +116,7 @@ pub fn resolve(store: &Store, cfg: &BotConfig, session: &str) -> Result Result<()> { } /// Records `system` (already handed to a new session) as that session's snapshot. +#[cfg(test)] pub fn adopt(store: &Store, cfg: &BotConfig, session: &str, system: String) -> Result { + adopt_with_prelude(store, cfg, session, system, None) +} + +/// New sessions retain their first-prompt instructions until the prompt is written. +pub fn adopt_new(store: &Store, cfg: &BotConfig, session: &str, system: String) -> Result { + adopt_with_prelude(store, cfg, session, system, Some(Prelude::New)) +} + +fn adopt_with_prelude( + store: &Store, + cfg: &BotConfig, + session: &str, + system: String, + prelude: Option, +) -> Result { let identity = Identity::of(cfg); let s = Snapshot { session: session.to_owned(), @@ -136,11 +164,31 @@ pub fn adopt(store: &Store, cfg: &BotConfig, session: &str, system: String) -> R system, system_identity: identity.clone(), announced: identity, + prelude, }; save(store, &cfg.id, &s)?; Ok(s) } +/// A fork inherits the instructions its parent actually heard, plus a thread intro. +pub fn fork(store: &Store, cfg: &BotConfig, parent: &Snapshot, session: &str) -> Result<()> { + let mut snapshot = parent.clone(); + session.clone_into(&mut snapshot.session); + snapshot.epoch = epoch(store, &cfg.id, session); + snapshot.prelude = Some(if parent.prelude == Some(Prelude::New) { Prelude::UnpromptedFork } else { Prelude::Fork }); + save(store, &cfg.id, &snapshot) +} + +/// A sent prompt clears only its own pending instructions, never another session's. +pub fn mark_prompted(store: &Store, bot_id: &str, sent: &Snapshot) -> Result<()> { + let Some(mut current) = load(store, bot_id, &sent.session) else { return Ok(()) }; + if current.epoch != sent.epoch || current.system != sent.system || current.prelude.is_none() { + return Ok(()); + } + current.prelude = None; + save(store, bot_id, ¤t) +} + /// The bot's instructions, rendered live. Reads memory files. pub fn render(store: &Store, cfg: &BotConfig) -> String { let name = cfg.name.trim(); diff --git a/host/tests/agent_allocation_e2e.rs b/host/tests/agent_allocation_e2e.rs new file mode 100644 index 000000000..44ffe757f --- /dev/null +++ b/host/tests/agent_allocation_e2e.rs @@ -0,0 +1,111 @@ +//! Interrupted allocations are recycled before another session is allocated. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +#[allow(dead_code)] +mod common; +#[path = "common/lifecycle.rs"] +mod lifecycle; + +use lifecycle::{Host, events, sid}; +use serde_json::json; +use std::time::Duration; + +#[tokio::test] +async fn routine_deadline_during_allocation_releases_uncertain_tools_before_retry() { + let host = Host::start().await; + let (bot, cwd) = host.bot("AllocationDeadline", json!({})).await; + // Keep the default no-close capability. A separate main lane pays startup + // before the routine's deadline, while its thread still allocates afresh. + let main = host.send(&bot, "prewarm the independent main lane", None).await; + let generation = events(&cwd)[0]["generation"].clone(); + std::fs::write(cwd.join("hang-new"), "").unwrap(); + let saved = host + .call( + "saveRoutine", + json!({"botId":bot,"name":"Setup timeout", + "instruction":"ok","triggers":[{"type":"webhook"}],"timeoutSeconds":2}), + ) + .await; + let routine = saved["routine"]["id"].as_str().unwrap(); + host.call("runRoutine", json!({"botId":bot,"id":routine})).await; + let allocation = + host.wait_event(&cwd, |event| event["event"] == "allocated" && event["session"] != sid(&main)).await; + assert_eq!(allocation["generation"], generation); + let completing = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let listing = host.call("routines", json!({"botId":bot})).await; + if let Some(run) = listing["runs"] + .as_array() + .unwrap() + .iter() + .find(|run| run["routineId"] == routine && run["status"] == "failed") + { + break run.clone(); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await; + let failed = completing.unwrap(); + // The main's pair and the unconfirmed routine allocation's pair remain + // until retrying the fresh thread forces safe process recycling. + host.wait_live(4, Duration::from_secs(15)).await; + std::fs::remove_file(cwd.join("hang-new")).unwrap(); + let root = failed["rootId"].as_str().unwrap(); + let replacement = host.send(&bot, "ok", Some(root)).await; + assert_ne!(sid(&replacement), allocation["session"].as_str().unwrap()); + let observed = events(&cwd); + let prompted = observed.iter().rev().find(|event| event["event"] == "prompted").unwrap(); + assert_ne!(prompted["generation"], generation); + assert_eq!(prompted["session"], sid(&replacement)); + host.wait_live_session(2, prompted["generation"].as_str().unwrap(), sid(&replacement), Duration::from_secs(15)) + .await; +} + +#[tokio::test] +async fn interrupted_nonresumable_allocation_keeps_confirmed_context_until_explicit_reset() { + let host = Host::start().await; + let (bot, cwd) = host.bot("NoResumeDeadline", json!({"load":false})).await; + let main = host.send(&bot, "main", None).await; + std::fs::write(cwd.join("hang-new"), "").unwrap(); + let saved = host + .call( + "saveRoutine", + json!({"botId":bot,"name":"Cancelled allocation", + "instruction":"ok","triggers":[{"type":"webhook"}],"timeoutSeconds":2}), + ) + .await; + let routine = saved["routine"]["id"].as_str().unwrap(); + host.call("runRoutine", json!({"botId":bot,"id":routine})).await; + host.wait_event(&cwd, |event| event["event"] == "allocated" && event["session"] != sid(&main)).await; + let completion = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let listing = host.call("routines", json!({"botId":bot})).await; + if listing["runs"] + .as_array() + .unwrap() + .iter() + .any(|run| run["routineId"] == routine && run["status"] == "failed") + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await; + completion.unwrap(); + let continued = host.send(&bot, "main continues", None).await; + assert_eq!(sid(&continued), sid(&main)); + std::fs::remove_file(cwd.join("hang-new")).unwrap(); + let failed = host.send_entry(&bot, "new thread", continued["id"].as_str()).await; + host.wait_entry(&bot, continued["id"].as_str(), |entry| { + entry["turn"] == failed["turn"] && entry["kind"] == "notice" && entry["data"]["style"] == "error" + }) + .await; + assert_eq!(events(&cwd).iter().filter(|event| event["event"] == "allocated").count(), 2); + host.call("newSession", json!({"botId":bot})).await; + let replacement = host.send(&bot, "new main", None).await; + assert_ne!(sid(&replacement), sid(&main)); + host.wait_live(2, Duration::from_secs(15)).await; +} diff --git a/host/tests/agent_fast_mode_lifecycle_e2e.rs b/host/tests/agent_fast_mode_lifecycle_e2e.rs new file mode 100644 index 000000000..f947bc0f8 --- /dev/null +++ b/host/tests/agent_fast_mode_lifecycle_e2e.rs @@ -0,0 +1,36 @@ +//! Fast mode changes are not session allocations. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +#[allow(dead_code)] +mod common; +#[path = "common/lifecycle.rs"] +mod lifecycle; + +use lifecycle::{Host, events, sid}; +use serde_json::{Value, json}; + +fn assert_speed(reply: &Value, on: bool) { + let expected = if on { " FAST=on" } else { " FAST=off" }; + assert!(reply["data"]["text"].as_str().unwrap().ends_with(expected), "{reply}"); +} + +#[tokio::test] +async fn fast_mode_changes_keep_nonresumable_main_and_thread_sessions_live() { + let host = Host::start().await; + let (bot, cwd) = host.bot("FastNoResume", json!({"fastMode":true,"load":false})).await; + host.call("updateBot", json!({"id":bot,"fastMode":true})).await; + let main = host.send(&bot, "main", None).await; + assert_speed(&main, true); + let root = main["id"].as_str().unwrap(); + let thread = host.send(&bot, "thread", Some(root)).await; + assert_speed(&thread, true); + host.call("updateBot", json!({"id":bot,"fastMode":false})).await; + let main_off = host.send(&bot, "main off", None).await; + let thread_off = host.send(&bot, "thread off", Some(root)).await; + assert_speed(&main_off, false); + assert_speed(&thread_off, false); + assert_eq!(sid(&main_off), sid(&main)); + assert_eq!(sid(&thread_off), sid(&thread)); + assert_eq!(events(&cwd).iter().filter(|event| event["event"] == "started").count(), 1); +} diff --git a/host/tests/agent_session_policy_e2e.rs b/host/tests/agent_session_policy_e2e.rs new file mode 100644 index 000000000..44c7aa33d --- /dev/null +++ b/host/tests/agent_session_policy_e2e.rs @@ -0,0 +1,52 @@ +//! Process ownership preserves the upstream allocation and recovery policy. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +#[allow(dead_code)] +mod common; +#[path = "common/lifecycle.rs"] +mod lifecycle; + +use lifecycle::{Host, events, sid}; +use serde_json::json; + +#[tokio::test] +async fn nonresumable_agents_keep_multiple_threads_and_accept_new_main_sessions() { + let host = Host::start().await; + let (bot, cwd) = host.bot("NoSessionCap", json!({"load":false,"fork":true})).await; + let main = host.send(&bot, "main", None).await; + let first = host.send(&bot, "first thread", main["id"].as_str()).await; + let next = host.send(&bot, "main continues", None).await; + let second = host.send(&bot, "second thread", next["id"].as_str()).await; + assert_ne!(sid(&first), sid(&second)); + assert_ne!(sid(&first), sid(&main)); + let fork_prompts = events(&cwd); + assert!(fork_prompts.iter().any(|event| event["event"] == "prompted" && event["forkIntro"] == true)); + host.call("newSession", json!({"botId":bot})).await; + let replacement = host.send(&bot, "fresh main", None).await; + assert_ne!(sid(&replacement), sid(&main)); + let continued = host.send(&bot, "first thread continues", main["id"].as_str()).await; + assert_eq!(sid(&continued), sid(&first)); + assert_eq!(events(&cwd).iter().filter(|event| event["event"] == "started").count(), 1); +} + +#[tokio::test] +async fn claude_style_missing_saved_sessions_recover_main_and_threads_with_a_notice() { + let mut host = Host::start().await; + let (bot, cwd) = host.bot("MissingTranscript", json!({})).await; + let main = host.send(&bot, "main", None).await; + let root = main["id"].as_str().unwrap(); + let thread = host.send(&bot, "thread", Some(root)).await; + host.restart().await; + std::fs::write(cwd.join("missing-load"), "").unwrap(); + let recovered_main = host.send(&bot, "recover main", None).await; + let recovered_thread = host.send(&bot, "recover thread", Some(root)).await; + assert_ne!(sid(&recovered_main), sid(&main)); + assert_ne!(sid(&recovered_thread), sid(&thread)); + for lane in [None, Some(root)] { + let entries = host.entries(&bot, lane).await; + assert!(entries.iter().any(|entry| entry["kind"] == "notice" + && entry["data"]["style"] == "divider" + && entry["data"]["text"].as_str().is_some_and(|text| text.contains("starting a fresh one")))); + } +} diff --git a/host/tests/agent_setup_e2e.rs b/host/tests/agent_setup_e2e.rs new file mode 100644 index 000000000..087898742 --- /dev/null +++ b/host/tests/agent_setup_e2e.rs @@ -0,0 +1,177 @@ +//! Failed session preparation must retain context and first-prompt instructions. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +#[allow(dead_code)] +mod common; +#[path = "common/lifecycle.rs"] +mod lifecycle; + +use lifecycle::{Host, events, sid}; +use serde_json::json; + +fn saved_session(host: &Host, bot: &str, root: Option<&str>) -> String { + let store = + rusqlite::Connection::open_with_flags(host.home.join("codync.db"), rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .unwrap(); + if let Some(root) = root { + let key = format!("lane.session.{bot}@{bot}/{root}"); + store.query_row("SELECT v FROM kv WHERE k=?1", [&key], |row| row.get(0)).unwrap() + } else { + store.query_row("SELECT session_id FROM bots WHERE id=?1", [bot], |row| row.get(0)).unwrap() + } +} + +#[tokio::test] +async fn retry_after_model_failure_keeps_the_new_session_and_delivers_its_profile_once() { + let host = Host::start().await; + for resumable in [true, false] { + let (bot, cwd) = host.bot(&format!("ModelRetry{resumable}"), json!({"load":resumable})).await; + host.call("updateBot", json!({"id":bot,"model":"fixture-model"})).await; + std::fs::write(cwd.join("fail-model"), "").unwrap(); + let failed = host.send_entry(&bot, "ok", None).await; + host.wait_entry(&bot, None, |entry| { + entry["turn"] == failed["turn"] && entry["kind"] == "notice" && entry["data"]["style"] == "error" + }) + .await; + let saved = saved_session(&host, &bot, None); + assert_ne!(saved, ""); + std::fs::remove_file(cwd.join("fail-model")).unwrap(); + let retried = host.send(&bot, "ok", None).await; + assert_eq!(sid(&retried), saved); + host.send(&bot, "ok again", None).await; + let prompts: Vec<_> = events(&cwd).into_iter().filter(|event| event["event"] == "prompted").collect(); + assert_eq!(prompts.len(), 2); + assert_eq!(prompts[0]["profile"], true, "initial instructions must survive setup failure"); + assert_eq!(prompts[1]["profile"], false, "completed instructions must not replay"); + } +} + +#[tokio::test] +async fn retry_after_fork_model_failure_keeps_the_thread_and_delivers_its_intro_once() { + let host = Host::start().await; + let (bot, cwd) = host.bot("ForkModelRetry", json!({"fork":true,"close":true})).await; + host.call("updateBot", json!({"id":bot,"model":"fixture-model"})).await; + let main = host.send(&bot, "ok", None).await; + let root = main["id"].as_str().unwrap(); + std::fs::write(cwd.join("fail-model"), "").unwrap(); + let failed = host.send_entry(&bot, "ok", Some(root)).await; + host.wait_entry(&bot, Some(root), |entry| { + entry["turn"] == failed["turn"] && entry["kind"] == "notice" && entry["data"]["style"] == "error" + }) + .await; + let saved = saved_session(&host, &bot, Some(root)); + assert_ne!(saved, sid(&main)); + std::fs::remove_file(cwd.join("fail-model")).unwrap(); + let retried = host.send(&bot, "ok", Some(root)).await; + assert_eq!(sid(&retried), saved); + host.send(&bot, "ok again", Some(root)).await; + let prompts: Vec<_> = + events(&cwd).into_iter().filter(|event| event["event"] == "prompted" && event["session"] == saved).collect(); + assert_eq!(prompts.len(), 2); + assert_eq!(prompts[0]["forkIntro"], true, "thread instructions must survive setup failure"); + assert_eq!(prompts[1]["threadIntro"], false); +} + +#[tokio::test] +async fn pending_thread_instructions_survive_other_lanes_and_host_restart() { + let mut host = Host::start().await; + let (bot, cwd) = host.bot("ThreadRestart", json!({"close":true})).await; + host.call("updateBot", json!({"id":bot,"model":"fixture-model"})).await; + let main = host.send(&bot, "ok", None).await; + let root = main["id"].as_str().unwrap(); + std::fs::write(cwd.join("fail-model"), "").unwrap(); + let failed = host.send_entry(&bot, "ok", Some(root)).await; + host.wait_entry(&bot, Some(root), |entry| { + entry["turn"] == failed["turn"] && entry["kind"] == "notice" && entry["data"]["style"] == "error" + }) + .await; + let saved = saved_session(&host, &bot, Some(root)); + std::fs::remove_file(cwd.join("fail-model")).unwrap(); + host.send(&bot, "main continues", None).await; + let prior = events(&cwd); + let main_prompt = prior.iter().rev().find(|event| event["event"] == "prompted").unwrap(); + assert_eq!(main_prompt["profile"], false); + assert_eq!(main_prompt["threadIntro"], false); + host.restart().await; + let resumed = host.send(&bot, "thread continues", Some(root)).await; + assert_eq!(sid(&resumed), saved); + let after = events(&cwd); + let thread_prompt = after.iter().rev().find(|event| event["event"] == "prompted").unwrap(); + assert_eq!(thread_prompt["profile"], true); + assert_eq!(thread_prompt["threadIntro"], true); +} + +#[tokio::test] +async fn routine_model_deadline_preserves_the_allocated_session_for_manual_retry() { + let host = Host::start().await; + let (bot, cwd) = host.bot("ModelDeadline", json!({})).await; + host.call("updateBot", json!({"id":bot,"model":"fixture-model"})).await; + let main = host.send(&bot, "prewarm the independent main lane", None).await; + std::fs::write(cwd.join("hang-model"), "").unwrap(); + let saved = host + .call( + "saveRoutine", + json!({"botId":bot,"name":"Model setup deadline","instruction":"ok", + "triggers":[{"type":"webhook"}],"timeoutSeconds":2}), + ) + .await; + let routine = saved["routine"]["id"].as_str().unwrap(); + host.call("runRoutine", json!({"botId":bot,"id":routine})).await; + let pending = + host.wait_event(&cwd, |event| event["event"] == "model_pending" && event["session"] != sid(&main)).await; + let failed = tokio::time::timeout(std::time::Duration::from_secs(15), async { + loop { + let listing = host.call("routines", json!({"botId":bot})).await; + if let Some(run) = listing["runs"] + .as_array() + .unwrap() + .iter() + .find(|run| run["routineId"] == routine && run["status"] == "failed") + { + break run.clone(); + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + }) + .await; + let failed = failed.unwrap(); + let root = failed["rootId"].as_str().unwrap(); + let identity = saved_session(&host, &bot, Some(root)); + assert_eq!(pending["session"], identity); + std::fs::remove_file(cwd.join("hang-model")).unwrap(); + let retried = host.send(&bot, "manual retry", Some(root)).await; + assert_eq!(sid(&retried), identity); + let observed = events(&cwd); + assert_eq!(observed.iter().filter(|event| event["event"] == "started").count(), 1); + assert_eq!(observed.iter().filter(|event| event["event"] == "allocated").count(), 2); + let prompt = observed.iter().find(|event| event["event"] == "prompted" && event["session"] == identity).unwrap(); + assert_eq!(prompt["profile"], true); + assert_eq!(prompt["threadIntro"], true); +} + +#[tokio::test] +async fn forking_before_the_parent_first_prompt_keeps_each_sessions_initial_profile() { + let host = Host::start().await; + let (bot, cwd) = host.bot("UnpromptedParent", json!({"fork":true,"close":true})).await; + host.call("updateBot", json!({"id":bot,"model":"fixture-model"})).await; + std::fs::write(cwd.join("fail-model"), "").unwrap(); + let failed = host.send_entry(&bot, "ok", None).await; + host.wait_entry(&bot, None, |entry| { + entry["turn"] == failed["turn"] && entry["kind"] == "notice" && entry["data"]["style"] == "error" + }) + .await; + let main = saved_session(&host, &bot, None); + std::fs::remove_file(cwd.join("fail-model")).unwrap(); + let root = failed["id"].as_str().unwrap(); + host.send(&bot, "thread first", Some(root)).await; + let parent = host.send(&bot, "parent first", None).await; + assert_eq!(sid(&parent), main); + let observed = events(&cwd); + let prompts: Vec<_> = observed.iter().filter(|event| event["event"] == "prompted").collect(); + assert_eq!(prompts.len(), 2); + assert_eq!(prompts[0]["profile"], true); + assert_eq!(prompts[0]["forkIntro"], true); + assert_eq!(prompts[1]["profile"], true); + assert_eq!(prompts[1]["threadIntro"], false); +} diff --git a/host/tests/common/lifecycle.rs b/host/tests/common/lifecycle.rs new file mode 100644 index 000000000..4473be1aa --- /dev/null +++ b/host/tests/common/lifecycle.rs @@ -0,0 +1,364 @@ +//! Persistent ACP host fixture shared by lifecycle and setup regressions. + +#![allow(dead_code)] + +use crate::common; +use serde::Deserialize; +use serde_json::{Value, json}; +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; + +#[derive(Clone, Debug, Deserialize)] +struct Descendant { + pid: u32, + role: String, + generation: String, + allocation: String, + session: Option, +} + +impl Descendant { + fn same_allocation(&self, other: &Self) -> bool { + self.role == other.role + && self.generation == other.generation + && self.allocation == other.allocation + && self.session == other.session + } +} + +type Inventory = Arc>>; + +struct Registration { + live: Inventory, + descendant: Descendant, +} + +impl Drop for Registration { + fn drop(&mut self) { + let key = (self.descendant.pid, self.descendant.allocation.clone()); + let mut live = self.live.lock().unwrap(); + if live.get(&key).is_some_and(|current| current.same_allocation(&self.descendant)) { + live.remove(&key); + } + } +} + +fn peer_closed(error: &std::io::Error) -> bool { + matches!( + error.kind(), + std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::NotConnected + | std::io::ErrorKind::UnexpectedEof + ) +} + +fn reserve_host_port() -> std::net::TcpListener { + // Outside the default ephemeral ranges on Linux, macOS and Windows CI, + // so a parallel witness listener or outgoing connection cannot claim it. + // Custom OS ranges and other processes can still collide during handoff; + // host startup failures remain fatal, without retrying the launched host. + const FIRST: u16 = 10_240; + const COUNT: u16 = 22_528; + for _ in 0..32 { + let offset = u16::try_from(uuid::Uuid::new_v4().as_u128() % u128::from(COUNT)).unwrap(); + let port = FIRST + offset; + match std::net::TcpListener::bind(("127.0.0.1", port)) { + Ok(reservation) => return reservation, + Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => {} + Err(error) => panic!("couldn't reserve lifecycle fixture port {port}: {error}"), + } + } + panic!("no available lifecycle fixture HTTP port after 32 candidates in 10240..32768"); +} + +pub struct Host { + child: Child, + pub home: PathBuf, + base: String, + token: String, + witness_port: u16, + live: Inventory, + witness_error: Arc>>, + witness_task: tokio::task::JoinHandle<()>, +} + +impl Drop for Host { + fn drop(&mut self) { + common::stop(&mut self.child); + self.witness_task.abort(); + let _ = std::fs::remove_dir_all(&self.home); + if !std::thread::panicking() { + self.assert_witness(); + } + } +} + +impl Host { + pub async fn restart(&mut self) { + common::stop(&mut self.child); + let port = self.base.rsplit(':').next().unwrap(); + self.child = Command::new(env!("CARGO_BIN_EXE_codync-host")) + .args(["serve", "--bind", "127.0.0.1", "--port", port]) + .env("CODYNC_CLOUD", "off") + .env("CODYNC_HOME", &self.home) + .stdout(Stdio::null()) + .stderr(std::fs::File::create(self.home.join("host-startup.log")).unwrap()) + .spawn() + .unwrap(); + self.wait_ready().await; + } + + pub async fn start() -> Self { + let home = std::env::temp_dir().join(format!("codync-lifecycle-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&home).unwrap(); + common::seed_memory_runtime(&home); + let witness = tokio::net::TcpListener::bind("127.0.0.1:0").await; + let witness = witness.unwrap(); + let witness_port = witness.local_addr().unwrap().port(); + let live = Arc::new(Mutex::new(BTreeMap::<(u32, String), Descendant>::new())); + let witness_error = Arc::new(Mutex::new(None)); + let reported_error = witness_error.clone(); + let observed = live.clone(); + let witness_task = tokio::spawn(async move { + while let Ok((stream, _)) = witness.accept().await { + let observed = observed.clone(); + let reported_error = reported_error.clone(); + tokio::spawn(async move { + let mut reader = BufReader::new(stream); + let mut greeting = String::new(); + let greeting_result = reader.read_line(&mut greeting).await; + match greeting_result { + Ok(0) => return, + Ok(_) => {} + Err(error) => { + if !peer_closed(&error) { + reported_error + .lock() + .unwrap() + .get_or_insert_with(|| format!("witness greeting failed: {:?}", error.kind())); + } + return; + } + } + let decoded = serde_json::from_str::(&greeting); + let descendant = match decoded { + Ok(descendant) => descendant, + Err(error) => { + reported_error + .lock() + .unwrap() + .get_or_insert_with(|| format!("invalid witness greeting: {:?}", error.classify())); + return; + } + }; + let key = (descendant.pid, descendant.allocation.clone()); + observed.lock().unwrap().insert(key.clone(), descendant.clone()); + let _registration = Registration { live: observed, descendant: descendant.clone() }; + let acknowledged = reader.get_mut().write_all(b"ready\n").await; + if let Err(error) = acknowledged { + if !peer_closed(&error) { + reported_error.lock().unwrap().get_or_insert_with(|| { + format!("witness acknowledgement failed: {:?}; {descendant:?}", error.kind()) + }); + } + return; + } + let mut remaining = Vec::new(); + let closed = reader.read_to_end(&mut remaining).await; + if let Err(error) = closed + && !peer_closed(&error) + { + reported_error + .lock() + .unwrap() + .get_or_insert_with(|| format!("witness read failed: {:?}; {descendant:?}", error.kind())); + } + }); + } + }); + let reservation = reserve_host_port(); + let port = reservation.local_addr().unwrap().port(); + let port_arg = port.to_string(); + let mut command = Command::new(env!("CARGO_BIN_EXE_codync-host")); + command + .args(["serve", "--bind", "127.0.0.1", "--port", &port_arg]) + .env("CODYNC_CLOUD", "off") + .env("CODYNC_HOME", &home) + .stdout(Stdio::null()) + .stderr(std::fs::File::create(home.join("host-startup.log")).unwrap()); + // Keep the availability probe until the configured child can be spawned. + drop(reservation); + let child = command.spawn().unwrap(); + let mut host = Self { + child, + home, + base: format!("http://127.0.0.1:{port}"), + token: String::new(), + witness_port, + live, + witness_error, + witness_task, + }; + host.wait_ready().await; + std::fs::read_to_string(host.home.join("token")).unwrap().trim().clone_into(&mut host.token); + host + } + + async fn wait_ready(&mut self) { + // CI starts several hosts together; binary hashing and login-shell PATH + // discovery compete for the runner before the HTTP server can respond. + let deadline = Instant::now() + Duration::from_secs(60); + let client = reqwest::Client::builder().timeout(Duration::from_secs(2)).build().unwrap(); + loop { + let response = client.get(format!("{}/health", self.base)).send().await; + if response.is_ok_and(|response| response.status().is_success()) { + return; + } + let exited = self.child.try_wait().unwrap(); + if exited.is_some() || Instant::now() >= deadline { + let log = std::fs::read_to_string(self.home.join("host-startup.log")).unwrap_or_default(); + panic!("host failed to become ready (exit={exited:?}): {log}"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + pub async fn call(&self, method: &str, body: Value) -> Value { + let res = reqwest::Client::new() + .post(format!("{}/api/{method}", self.base)) + .bearer_auth(&self.token) + .json(&body) + .send() + .await; + let res = res.unwrap(); + let status = res.status(); + let decoded = res.json().await; + let value: Value = decoded.unwrap(); + assert!(status.is_success(), "{method}: {value}"); + value + } + + pub async fn bot(&self, name: &str, mut config: Value) -> (String, PathBuf) { + let cwd = self.home.join(name); + std::fs::create_dir_all(&cwd).unwrap(); + config["witnessPort"] = json!(self.witness_port); + std::fs::write(cwd.join("fixture.json"), config.to_string()).unwrap(); + let script = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/lifecycle_agent.py"); + let response = self + .call( + "createBot", + json!({"name":name,"backend":"custom","cwd":cwd, + "command":common::python_agent(&script),"permission":"ask","notify":false}), + ) + .await; + (response["bot"]["id"].as_str().unwrap().to_owned(), cwd) + } + + pub async fn send_entry(&self, bot: &str, text: &str, thread: Option<&str>) -> Value { + let response = self + .call( + "send", + json!({"botId":bot,"text":text,"threadId":thread, + "clientNonce":uuid::Uuid::new_v4().to_string()}), + ) + .await; + response["entry"].clone() + } + + pub async fn entries(&self, bot: &str, thread: Option<&str>) -> Vec { + let response = if let Some(root) = thread { + self.call("thread", json!({"botId":bot,"rootId":root})).await + } else { + self.call("history", json!({"botId":bot,"limit":500})).await + }; + response["entries"].as_array().unwrap().clone() + } + + pub async fn wait_entry(&self, bot: &str, thread: Option<&str>, pred: impl Fn(&Value) -> bool) -> Value { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let entries = self.entries(bot, thread).await; + if let Some(entry) = entries.iter().rev().find(|e| pred(e)) { + return entry.clone(); + } + assert!(Instant::now() < deadline, "missing expected entry: {entries:?}"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + pub async fn send(&self, bot: &str, text: &str, thread: Option<&str>) -> Value { + let entry = self.send_entry(bot, text, thread).await; + self.wait_entry(bot, thread, |e| { + e["turn"] == entry["turn"] && e["kind"] == "agent" && e["data"]["final"] == true + }) + .await + } + + pub async fn wait_event(&self, cwd: &Path, pred: impl Fn(&Value) -> bool) -> Value { + let deadline = Instant::now() + Duration::from_secs(15); + loop { + self.assert_witness(); + let observed = events(cwd); + if let Some(event) = observed.iter().rev().find(|event| pred(event)) { + return event.clone(); + } + assert!(Instant::now() < deadline, "missing expected fixture phase: {observed:?}"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + pub async fn wait_live(&self, expected: usize, budget: Duration) { + self.wait_inventory(expected, None, budget).await; + } + + pub async fn wait_live_session(&self, expected: usize, generation: &str, session: &str, budget: Duration) { + self.wait_inventory(expected, Some((generation, session)), budget).await; + } + + async fn wait_inventory(&self, expected: usize, identity: Option<(&str, &str)>, budget: Duration) { + // Starting multiple Python generations is slower on loaded Windows runners. + let budget = if cfg!(windows) && budget <= Duration::from_secs(5) { budget * 3 } else { budget }; + let deadline = Instant::now() + budget; + loop { + self.assert_witness(); + let inventory = self.live.lock().unwrap().clone(); + let matches = identity.is_none_or(|(generation, session)| { + inventory.values().all(|descendant| { + descendant.generation == generation && descendant.session.as_deref() == Some(session) + }) + }); + if inventory.len() == expected && matches { + break; + } + assert!( + Instant::now() < deadline, + "expected {expected} tool descendants for {identity:?}, saw {}: {inventory:?}", + inventory.len() + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + fn assert_witness(&self) { + let error = self.witness_error.lock().unwrap(); + assert!(error.is_none(), "unexpected witness error: {error:?}"); + } +} + +pub fn sid(reply: &Value) -> &str { + reply["data"]["text"].as_str().unwrap().split_whitespace().next().unwrap().strip_prefix("SID=").unwrap() +} + +pub fn events(cwd: &Path) -> Vec { + std::fs::read_to_string(cwd.join("events.jsonl")) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect() +} diff --git a/host/tests/common/mod.rs b/host/tests/common/mod.rs index e6ed28f9c..d8b1b8e70 100644 --- a/host/tests/common/mod.rs +++ b/host/tests/common/mod.rs @@ -62,15 +62,19 @@ pub async fn serve(home: &Path) -> (Child, String) { /// Stops a test host the way launchd does (SIGTERM), so it shuts down its own /// children: a SIGKILL would orphan whatever it had running, such as a harness -/// sign-in check (`cursor-agent status` then spins forever). Kills it after 5 s. +/// sign-in check (`cursor-agent status` then spins forever). Allows the agent's ten-second graceful shutdown, then kills it after 15 s. +/// Windows uses its owned child handle directly. pub fn stop(child: &mut Child) { - let _ = Command::new("kill").arg(child.id().to_string()).status(); - let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline { - if let Ok(Some(_)) = child.try_wait() { - return; + #[cfg(not(windows))] + { + let _ = Command::new("kill").arg(child.id().to_string()).status(); + let deadline = Instant::now() + Duration::from_secs(15); + while Instant::now() < deadline { + if let Ok(Some(_)) = child.try_wait() { + return; + } + std::thread::sleep(Duration::from_millis(20)); } - std::thread::sleep(Duration::from_millis(20)); } let _ = child.kill(); let _ = child.wait(); diff --git a/host/tests/fixtures/lifecycle_agent.py b/host/tests/fixtures/lifecycle_agent.py new file mode 100644 index 000000000..e97b28e51 --- /dev/null +++ b/host/tests/fixtures/lifecycle_agent.py @@ -0,0 +1,254 @@ +"""Persistent ACP sessions, independently observable tool trees and load failures.""" + +import json +import os +from pathlib import Path +import signal +import socket +import subprocess +import sys +import threading +import time +import uuid + +if len(sys.argv) > 1: + # TCP EOF is an exit acknowledgement even when the host uses SIGKILL. + config = json.loads(Path("fixture.json").read_text()) + witness = socket.create_connection(("127.0.0.1", config["witnessPort"])) + identity = json.loads(sys.argv[2]) + witness.sendall((json.dumps({"pid": os.getpid(), "role": sys.argv[1], **identity}) + "\n").encode()) + if witness.makefile("rb").readline() != b"ready\n": + raise RuntimeError("witness did not acknowledge descendant") + leaf = None + if sys.argv[1] == "worker": + leaf = subprocess.Popen([sys.executable, "-u", __file__, "leaf", sys.argv[2]], stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE if config.get("waitForDescendants", False) else sys.stdout, + stderr=sys.stderr, text=True) + if config.get("waitForDescendants", False) and leaf.stdout.readline() != "ready\n": + raise RuntimeError("leaf did not acknowledge readiness") + if config.get("waitForDescendants", False): + print("ready", flush=True) + + def stop(*_): + if leaf is not None: + leaf.terminate() + leaf.wait(timeout=5) + sys.exit(0) + + signal.signal(signal.SIGTERM, stop) + if leaf is not None: + sys.stdin.readline() + stop() + else: + while True: + time.sleep(1) + +config = json.loads(Path("fixture.json").read_text()) +generation = str(uuid.uuid4()) +saved_file = Path("sessions.json") +saved = json.loads(saved_file.read_text()) if saved_file.exists() else {} +live = {} +fast_options = {} +workers = {} +background = [] +lock = threading.RLock() +cancel = threading.Event() +permission = threading.Event() + + +def send(value): + with lock: + print(json.dumps({"jsonrpc": "2.0", **value}), flush=True) + + +def event(kind, sid=None, **fields): + with lock, Path("events.jsonl").open("a") as output: + output.write(json.dumps({"event": kind, "generation": generation, + "session": sid, "live": len(live), **fields}) + "\n") + + +def persist(): + temp = saved_file.with_suffix(".tmp") + temp.write_text(json.dumps(saved)) + temp.replace(saved_file) + + +def session_options(sid): + if not config.get("fastMode", False): + return {} + return {"configOptions": [{ + "id": "fast", "name": "Fast mode", "type": "select", + "currentValue": fast_options.get(sid, "off"), + "options": [{"value": "on", "name": "On"}, {"value": "off", "name": "Off"}], + }]} + + +def spawn_worker(sid): + allocation = str(uuid.uuid4()) + identity = json.dumps({"generation": generation, "allocation": allocation, "session": sid}) + worker = subprocess.Popen([sys.executable, __file__, "worker", identity], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE if config.get("waitForDescendants", False) else None, + text=True) + if config.get("waitForDescendants", False) and worker.stdout.readline() != "ready\n": + raise RuntimeError("worker did not acknowledge readiness") + return worker, allocation + + +def allocate(sid, params): + # Deliberately leak resources on repeat load unless the client closes first. + if sid in workers: + event("duplicate_allocation", sid) + worker, allocation = spawn_worker(sid) + workers.setdefault(sid, []).append(worker) + live[sid] = params.get("_meta", {}) + fast_options[sid] = "off" + event("allocated", sid, allocation=allocation, + computerTool=any(server["name"] == "computer" for server in params.get("mcpServers", [])), + memoryLane=next((item["value"] for server in params.get("mcpServers", []) + for item in server.get("env", []) if item["name"] == "CODYNC_MEMORY_LANE"), None)) + + +def prompt(message): + sid = message["params"]["sessionId"] + text = message["params"]["prompt"][0]["text"] + if "BACKGROUND" in text: + worker, _ = spawn_worker(sid) + background.append(worker) + event("background_started", sid) + if "BLOCK" in text: + event("blocked", sid) + while not Path("release").exists() and not cancel.wait(.01): + pass + if "PERMISSION" in text: + permission.clear() + send({"id": 900, "method": "session/request_permission", "params": { + "sessionId": sid, "toolCall": {"toolCallId": "permission", "title": "Test approval", "kind": "execute"}, + "options": [{"optionId": "allow", "name": "Allow", "kind": "allow_once"}]}}) + event("permission", sid) + permission.wait() + if "COMPACT" in text: + send({"method": "session/update", "params": {"sessionId": sid, "update": { + "sessionUpdate": "compaction_update", "status": "completed", "compactionId": str(uuid.uuid4())}}}) + with lock: + saved[sid].append(text) + persist() + summary = f"SID={sid} COUNT={len(saved[sid])} {text.splitlines()[-1]}" + if config.get("fastMode", False): + summary += f" FAST={fast_options[sid]}" + send({"method": "session/update", "params": {"sessionId": sid, "update": { + "sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": summary}}}}) + event("prompted", sid, profile="" in text, threadIntro="[Thread]" in text, + forkIntro="starts from everything said so far" in text) + send({"id": message["id"], "result": {"stopReason": "cancelled" if cancel.is_set() else "end_turn"}}) + + +event("started") +for line in sys.stdin: + message = json.loads(line) + method = message.get("method") + params = message.get("params", {}) + if method is None: + if message.get("id") == 900: + permission.set() + continue + rid = message.get("id") + if method == "initialize": + caps = {"loadSession": config.get("load", True), "sessionCapabilities": {}} + if config.get("close", False): + caps["sessionCapabilities"]["close"] = {} + if config.get("fork", False): + caps["sessionCapabilities"]["fork"] = {} + if config.get("claude", False): + caps["_meta"] = {"claudeCode": {}} + send({"id": rid, "result": {"protocolVersion": 1, "agentCapabilities": caps}}) + elif method == "session/new": + if Path("opaque-new").exists(): + event("new_failed") + send({"id": rid, "error": {"code": -32603, "message": "Internal error", + "data": {"message": "Failed to initialize session services"}}}) + continue + sid = str(uuid.uuid4()) + saved[sid] = [] + persist() + allocate(sid, params) + if Path("hang-new").exists(): + continue + if Path("fail-new").exists(): + send({"id": rid, "error": {"code": -32001, "message": "allocation failed after tools started"}}) + else: + send({"id": rid, "result": {"sessionId": sid, **session_options(sid)}}) + elif method == "session/fork": + parent = params["sessionId"] + if parent not in live: + send({"id": rid, "error": {"code": -32602, "message": "parent must be live"}}) + else: + sid = str(uuid.uuid4()) + saved[sid] = saved[parent].copy() + persist() + allocate(sid, params) + event("forked", sid, parent=parent) + returned = {"empty": "", "parent": parent}.get(config.get("forkResponse"), sid) + send({"id": rid, "result": {"sessionId": returned, **session_options(sid)}}) + elif method == "session/load": + sid = params["sessionId"] + if Path("crash-load").exists(): + event("load_crashed", sid) + os._exit(1) + elif Path("opaque-load").exists(): + event("load_failed", sid) + send({"id": rid, "error": {"code": -32603, "message": "Internal error", + "data": {"message": "Failed to initialize session services"}}}) + elif Path("missing-load").exists(): + event("load_failed", sid) + send({"id": rid, "error": {"code": -32002, "message": f"Resource not found: {sid}"}}) + elif Path("broken-load").exists(): + event("load_failed", sid) + send({"id": rid, "error": {"code": -32603, "message": "Internal error", + "data": {"message": f'Failed to load session "{sid}": corrupt session storage'}}}) + elif Path("fail-load").exists(): + event("load_failed", sid) + send({"id": rid, "error": {"code": -32602, "message": "saved session is unavailable"}}) + elif Path("auth-load").exists(): + send({"id": rid, "error": {"code": -32000, "message": "authentication required"}}) + elif sid not in saved: + send({"id": rid, "error": {"code": -32602, "message": "missing saved session"}}) + else: + allocate(sid, params) + send({"method": "session/update", "params": {"sessionId": sid, "update": { + "sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "REPLAY"}}}}) + event("loaded", sid) + send({"id": rid, "result": session_options(sid)}) + elif method == "session/close": + event("closing", params["sessionId"]) + if config.get("closeHangs", False): + continue + if config.get("closeFails", False): + send({"id": rid, "error": {"code": -32001, "message": "close failed"}}) + else: + sid = params["sessionId"] + for worker in workers.pop(sid, []): + worker.stdin.write("stop\n") + worker.stdin.flush() + worker.wait(timeout=5) + live.pop(sid, None) + fast_options.pop(sid, None) + event("closed", sid) + send({"id": rid, "result": {}}) + elif method == "session/prompt": + cancel.clear() + threading.Thread(target=prompt, args=(message,)).start() + elif method == "session/cancel": + cancel.set() + permission.set() + elif method in ("session/set_model", "session/set_config_option"): + if Path("hang-model").exists(): + event("model_pending", params["sessionId"]) + elif Path("fail-model").exists(): + event("model_failed", params["sessionId"]) + send({"id": rid, "error": {"code": -32001, "message": "transient model selection failure"}}) + else: + event("model", params["sessionId"]) + if config.get("fastMode", False) and params.get("configId") == "fast": + fast_options[params["sessionId"]] = params["value"] + send({"id": rid, "result": session_options(params["sessionId"])}) diff --git a/host/tests/fixtures/process_agent.py b/host/tests/fixtures/process_agent.py new file mode 100644 index 000000000..ad4b936bb --- /dev/null +++ b/host/tests/fixtures/process_agent.py @@ -0,0 +1,19 @@ +"""Keep two descendant generations holding ACP stdout after the leader exits.""" + +import json +import subprocess +import sys +import time + +if len(sys.argv) > 1: + if sys.argv[1] == "child": + subprocess.Popen([sys.executable, "-u", __file__, "leaf"], stdin=subprocess.DEVNULL, stdout=sys.stdout, stderr=sys.stderr) + if sys.argv[1] == "leaf": + print(json.dumps({"jsonrpc": "2.0", "method": "ready"}), flush=True) + time.sleep(60) +else: + subprocess.Popen([sys.executable, "-u", __file__, "child"], stdin=subprocess.DEVNULL, stdout=sys.stdout, stderr=sys.stderr) + for line in sys.stdin: + message = json.loads(line) + if message.get("method") == "exit": + break