fix(supervisor): resume a conversation only once Claude Code has it on disk

crates/botloftd/src/chat/events.rs

@@ -38,7 +38,7 @@ pub(super) fn apply(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value
     let from_subagent = !event["parent_tool_use_id"].is_null();
     match kind {
         "system" if subtype == Some("init") => {
-            session(daemon, bot, event);
+            init(daemon, bot, event);
             daemon.supervisor.turn_began(bot, generation);
         }
         "system" if subtype == Some("background_tasks_changed") => {
@@ -72,20 +72,8 @@ fn agent_count(event: &Value) -> u32 {
     u32::try_from(agents).unwrap_or(u32::MAX)
 }
 
-/// Every turn starts with the session id; the next start resumes it.
-fn session(daemon: &Daemon, bot: &BotId, event: &Value) {
-    let Some(session) = event["session_id"].as_str() else {
-        return;
-    };
-    let store = daemon.store();
-    let known = store.session_id(bot).ok().flatten();
-    if known.as_deref() != Some(session)
-        && let Err(err) = store.set_session_id(bot, Some(session))
-    {
-        warn!(bot = %bot, "could not save the session id: {err}");
-    }
-    drop(store);
-    daemon.supervisor.remember_session(bot, session);
+/// Every turn starts with the mode and the model in use (spec 7.4).
+fn init(daemon: &Daemon, bot: &BotId, event: &Value) {
     if let Some(mode) = event["permissionMode"].as_str() {
         service::modes::reported(daemon, bot, mode);
     }
@@ -211,6 +199,7 @@ fn failed_turn(daemon: &Daemon, bot: &BotId, generation: u64, error: &str, event
 
 /// The bot began the turn for a message the courier wrote (spec 9.1).
 fn replay(daemon: &Daemon, bot: &BotId, event: &Value) {
+    super::session::began(daemon, bot, event);
     let Some(uuid) = event["uuid"].as_str() else {
         return;
     };
@@ -303,6 +292,12 @@ fn rate_limit(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) {
 }
 
 fn result(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) {
+    // No turn began, so none ended: nothing of this is for the chat.
+    if super::session::missing(event) {
+        debug!(bot = %bot, "Claude Code does not have the conversation to resume");
+        daemon.supervisor.session_missing(bot, generation);
+        return;
+    }
     let failed = event["is_error"].as_bool() == Some(true);
     let error = failed.then(|| {
         event["terminal_reason"]

crates/botloftd/src/chat/mod.rs

@@ -4,6 +4,7 @@
 
 mod events;
 pub(crate) mod items;
+mod session;
 
 use botloft_core::ids::BotId;
 use tracing::debug;

crates/botloftd/src/chat/session.rs

@@ -0,0 +1,69 @@
+//! What the stream says about the conversation on Claude Code's side (spec
+//! 7.3): when it is on disk and can be resumed, and when the one a process
+//! was told to resume is not there.
+
+use botloft_core::ids::BotId;
+use serde_json::Value;
+use tracing::warn;
+
+use crate::state::Daemon;
+
+/// How the error of a missing conversation begins (seen with 2.1.284,
+/// spec 19); the session id follows.
+const NO_CONVERSATION: &str = "No conversation found";
+
+/// A turn's message came back, with the session id like every event. Claude
+/// Code has the conversation on disk from here on, not from `system/init`,
+/// so this is what the next start resumes.
+pub(super) fn began(daemon: &Daemon, bot: &BotId, event: &Value) {
+    let Some(session) = event["session_id"].as_str() else {
+        return;
+    };
+    // Only the bot's own turns: a subagent works inside one of them.
+    if !event["parent_tool_use_id"].is_null() {
+        return;
+    }
+    let store = daemon.store();
+    let known = store.session_id(bot).ok().flatten();
+    if known.as_deref() != Some(session)
+        && let Err(err) = store.set_session_id(bot, Some(session))
+    {
+        warn!(bot = %bot, "could not save the session id: {err}");
+    }
+    drop(store);
+    daemon.supervisor.remember_session(bot, session);
+}
+
+/// Whether this `result` is Claude Code saying it could not find the
+/// conversation to resume. It comes with no turn before it, and the process
+/// exits right after.
+pub(super) fn missing(event: &Value) -> bool {
+    event["subtype"] == "error_during_execution"
+        && event["errors"].as_array().is_some_and(|errors| {
+            errors
+                .iter()
+                .filter_map(Value::as_str)
+                .any(|error| error.starts_with(NO_CONVERSATION))
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::*;
+
+    fn failed(subtype: &str, error: &str) -> Value {
+        json!({ "type": "result", "subtype": subtype, "is_error": true, "errors": [error] })
+    }
+
+    #[test]
+    fn only_a_conversation_that_was_not_found_counts_as_missing() {
+        let not_found = "No conversation found with session ID: 7de9b8c8";
+        assert!(missing(&failed("error_during_execution", not_found)));
+        assert!(!missing(&failed("error_during_execution", "other")));
+        assert!(!missing(&failed("error_max_turns", not_found)));
+        let done = json!({ "type": "result", "subtype": "success", "is_error": false });
+        assert!(!missing(&done));
+    }
+}

crates/botloftd/src/supervisor/mod.rs

@@ -7,6 +7,7 @@
 
 mod reconcile;
 mod relaunch;
+mod session;
 mod settings;
 mod sign_in;
 mod slot;
@@ -61,11 +62,13 @@ struct Inner {
     slots: HashMap<BotId, Slot>,
     /// Token hash of each running generation -> its bot.
     tokens: HashMap<String, (BotId, u64)>,
-    /// The conversation each bot resumes, as far as this run knows; the
+    /// The conversation each bot resumes: one Claude Code has on disk. The
     /// database has it too (spec 7.3).
     sessions: HashMap<BotId, String>,
     claude: ClaudeStatus,
     sign_in: sign_in::SignIn,
+    /// Conversations a new one took the place of, still in the database.
+    replaced: Vec<(BotId, String)>,
 }
 
 #[derive(Default)]

crates/botloftd/src/supervisor/reconcile.rs

@@ -60,6 +60,8 @@ impl Supervisor {
                 self.reconcile_bot(&mut inner, &daemon, crew, bot, now);
             }
         }
+        drop(inner);
+        self.forget_replaced(&daemon);
     }
 
     fn reconcile_bot(

crates/botloftd/src/supervisor/session.rs

@@ -0,0 +1,45 @@
+//! Which conversation a bot's next process resumes (spec 7.3). Claude Code
+//! has nothing on disk for a conversation before a turn began in it, so
+//! only one that had a turn is remembered; one that Claude Code lost, or
+//! that a new one took the place of, is forgotten.
+
+use botloft_core::ids::BotId;
+use tracing::warn;
+
+use super::Supervisor;
+use crate::state::Daemon;
+
+impl Supervisor {
+    /// A turn began in `session`: Claude Code has it on disk from here on,
+    /// and the next start resumes it.
+    pub(crate) fn remember_session(&self, bot: &BotId, session: &str) {
+        self.lock().sessions.insert(bot.clone(), session.to_owned());
+    }
+
+    /// Claude Code does not have the conversation this process was told to
+    /// resume, and is about to exit. The next process starts a new one,
+    /// however long this one takes to go.
+    pub(crate) fn session_missing(&self, bot: &BotId, generation: u64) {
+        self.with_current(bot, generation, |_, slot| slot.fresh_next = true);
+    }
+
+    /// Takes the conversations that new ones replaced out of the database,
+    /// so a daemon that restarts does not resume them. Called without the
+    /// supervisor lock (lock order, see `mod.rs`).
+    pub(super) fn forget_replaced(&self, daemon: &Daemon) {
+        let replaced = std::mem::take(&mut self.lock().replaced);
+        if replaced.is_empty() {
+            return;
+        }
+        let store = daemon.store();
+        for (bot, session) in replaced {
+            // A turn of the new conversation may have been saved already.
+            let stored = store.session_id(&bot).ok().flatten();
+            if stored.as_deref() == Some(session.as_str())
+                && let Err(err) = store.set_session_id(&bot, None)
+            {
+                warn!(bot = %bot, "could not forget the old session id: {err}");
+            }
+        }
+    }
+}

crates/botloftd/src/supervisor/spawn.rs

@@ -40,6 +40,7 @@ impl Supervisor {
             slots,
             tokens,
             sessions,
+            replaced,
             claude,
             ..
         } = inner;
@@ -62,7 +63,13 @@ impl Supervisor {
             .and_then(|(spec, launch)| Ok((self.runtime.spawn(spec)?, launch)));
         match launched {
             Ok((process, launch)) => {
-                sessions.insert(bot.id.clone(), launch.session.clone());
+                // A new conversation is on disk only once it has a turn
+                // (spec 7.3); the one it takes the place of is left behind.
+                if !launch.resumed
+                    && let Some(old) = sessions.remove(&bot.id)
+                {
+                    replaced.push((bot.id.clone(), old));
+                }
                 slot.generation = Some(generation);
                 slot.restart_at = None;
                 slot.fresh_next = false;
@@ -130,7 +137,7 @@ impl Supervisor {
 
         match slot.stop.take() {
             Some(StopIntent::Restart { fresh }) => {
-                slot.fresh_next = fresh;
+                slot.fresh_next |= fresh;
                 slot.restart_at = Some(Instant::now());
             }
             Some(StopIntent::Halt(state)) => {
@@ -160,7 +167,6 @@ impl Supervisor {
 struct Launch {
     token_hash: String,
     resumed: bool,
-    session: String,
 }
 
 /// Command line and environment of spec 7.4, with a fresh bot token.
@@ -195,7 +201,7 @@ fn launch_spec(
     .map(OsString::from)
     .collect();
     args.push(if resumed { "--resume" } else { "--session-id" }.into());
-    args.push(session.clone().into());
+    args.push(session.into());
     for arg in [
         "--setting-sources",
         "project,local",
@@ -248,7 +254,6 @@ fn launch_spec(
         Launch {
             token_hash: TokenHash::of(&token).to_hex(),
             resumed,
-            session,
         },
     ))
 }

crates/botloftd/src/supervisor/turns.rs

@@ -11,7 +11,12 @@ use super::{Inner, Supervisor};
 
 impl Supervisor {
     /// Runs `change` on the slot of `bot` if `generation` is still running.
-    fn with_current(&self, bot: &BotId, generation: u64, change: impl FnOnce(&Self, &mut Slot)) {
+    pub(super) fn with_current(
+        &self,
+        bot: &BotId,
+        generation: u64,
+        change: impl FnOnce(&Self, &mut Slot),
+    ) {
         let mut inner = self.lock();
         let Inner { slots, .. } = &mut *inner;
         if let Some(slot) = slots.get_mut(bot)
@@ -22,12 +27,6 @@ impl Supervisor {
         }
     }
 
-    /// The conversation the bot is in, from `system/init`; the next start
-    /// resumes it (spec 7.3).
-    pub(crate) fn remember_session(&self, bot: &BotId, session: &str) {
-        self.lock().sessions.insert(bot.clone(), session.to_owned());
-    }
-
     /// The process lived through its first moments: it is ready.
     pub(crate) fn ready(&self, bot: &BotId, generation: u64) {
         self.with_current(bot, generation, |supervisor, slot| {

crates/botloftd/tests/bot_models.rs

@@ -45,7 +45,7 @@ async fn an_idle_bot_restarts_on_its_new_model_in_the_same_conversation() {
     let s = setup().await;
     let first = s.runtime.process(1).await;
     s.until(BotState::Idle).await;
-    let session = arg_after(&first, "--session-id").expect("session");
+    let session = s.turn(&first).await;
 
     set_model(&s, BotModel::Haiku);
     let second = s.runtime.process(2).await;

crates/botloftd/tests/common/stream.rs

@@ -18,6 +18,34 @@ pub fn replay(line: &Value) -> Value {
     })
 }
 
+/// The replay that opens a turn in `session`. From here on Claude Code has
+/// the conversation on disk and it can be resumed (spec 7.3).
+pub fn began(session: &str) -> Value {
+    json!({
+        "type": "user",
+        "isReplay": true,
+        "uuid": "u",
+        "session_id": session,
+        "message": { "role": "user", "content": "hi" },
+        "parent_tool_use_id": null,
+    })
+}
+
+/// What Claude Code 2.1.284 prints, before it exits, when the conversation
+/// it was told to resume is not on disk (spec 19).
+pub fn no_conversation(session: &str) -> Value {
+    json!({
+        "type": "result",
+        "subtype": "error_during_execution",
+        "is_error": true,
+        "duration_ms": 0,
+        "num_turns": 0,
+        "session_id": session,
+        "total_cost_usd": 0,
+        "errors": [format!("No conversation found with session ID: {session}")],
+    })
+}
+
 pub fn text(text: &str) -> Value {
     json!({
         "type": "assistant",

crates/botloftd/tests/common/supervised.rs

@@ -13,7 +13,7 @@ use botloftd::state::Daemon;
 use botloftd::supervisor;
 use serde_json::json;
 
-use super::{new_daemon, test_settings};
+use super::{new_daemon, stream, test_settings};
 
 pub struct Setup {
     pub daemon: Arc<Daemon>,
@@ -86,6 +86,25 @@ impl Setup {
             .write_message(&self.bot, format!("{line}\n").into())
             .expect("running")
     }
+
+    /// One whole turn in the conversation `process` is in. Returns its
+    /// session id, which Claude Code now has on disk (spec 7.3).
+    pub async fn turn(&self, process: &FakeProcess) -> String {
+        let session = session_of(process);
+        self.message("hi");
+        process.emit(stream::init(&session)).await;
+        process.emit(stream::began(&session)).await;
+        process.emit(stream::result(false)).await;
+        self.until(BotState::Idle).await;
+        session
+    }
+}
+
+/// The conversation the process started or resumed.
+pub fn session_of(process: &FakeProcess) -> String {
+    arg_after(process, "--session-id")
+        .or_else(|| arg_after(process, "--resume"))
+        .expect("a session")
 }
 
 /// The value after `flag` on the process's command line.