fix(host): own agent processes and preserve session setup

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` |

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
 

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.

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.

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<std::result::Result<Value, Value>>,
+    id: i64,
+    pending: Pending,
+    rx: Option<oneshot::Receiver<std::result::Result<Value, Value>>>,
 }
 
 impl Answer {
-    pub async fn response(self) -> Result<Value> {
-        match self.rx.await {
+    pub async fn response(mut self) -> Result<Value> {
+        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<Option<ChildStdin>>,
     next_id: AtomicI64,
     pending: Pending,
-    /// `None` once stopped (handed to [`reap`]).
-    child: tokio::sync::Mutex<Option<Child>>,
+    child: Arc<tokio::sync::Mutex<Process>>,
     /// The process (and, on Unix, process group) leader.
     pid: Option<u32>,
 }
@@ -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<Value> {
-        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<Acp>, mpsc::UnboundedReceiver<Incoming>) {
+        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<Incoming>) {
+        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!([

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");

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");

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<Hub>, 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<String, Instant>,
     started: Instant,
+    allocation_uncertain: bool,
+    /// Allocated sessions whose model setup has not yet completed.
+    pending_models: HashMap<String, Value>,
     /// Each live session's Fast mode option, as the agent last listed it.
     fast_options: HashMap<String, Value>,
 }
@@ -216,15 +218,11 @@ struct Actor {
     lane: Lane,
     /// The session the current turn prompts.
     turn_session: Option<String>,
-    /// Told on the first message of a new thread session: what the thread is about.
-    thread_intro: Option<String>,
     active_group: Option<GroupTurn>,
     active_routine: Option<String>,
     routine_deadline: Option<Instant>,
     routine_timed_out: bool,
     routine_completion: Option<RoutineCompletion>,
-    /// 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<String, String>,
     keeper: mpsc::UnboundedSender<memory::KeeperEvent>,