atlas-agent-manager: the six ATL-212 children — connect once, free the process, keep the turn id

ARCHITECTURE.md

@@ -167,7 +167,7 @@ Beyond `prompt` / `cancel` / `authenticate`, **every optional behaviour is capab
 
 Three layers sit between the connection and the IPC surface:
 
-- **`AgentManager`** (`crates/atlas-agent-manager`) owns who is connected and which sessions are open on them — ported from Zed's `AgentConnectionStore`. Three behaviours worth knowing: a second connect request while the first is still connecting *joins* it rather than starting a second process; a failed connection does not stick (the entry records the error for waiters, then goes, so the next request retries instead of replaying the failure forever); and a version bump drops the connection, because the running process is on the old binary.
+- **`AgentManager`** (`crates/atlas-agent-manager`) owns who is connected and which sessions are open on them — ported from Zed's `AgentConnectionStore`. Three behaviours worth knowing: a second connect request while the first is still connecting *joins* it rather than starting a second process; a failed connection does not stick (the entry records the error for waiters, then goes, so the next request retries instead of replaying the failure forever); and a version bump drops the connection, because the running process is on the old binary. Every eviction path also forgets that agent's sessions and cancels any connect still in flight — a session pins the connection `Arc`, so one left behind keeps a child alive that nothing, including the shutdown sweep, can reach.
 - **`AgentHost`** (`src-tauri/src/commands/agent_host.rs`) is what `commands/agents.rs` talks to. It holds the three things the ported stack deliberately does not do: the **identity map** between the frontend's per-spawn `AgentId`/`SessionKey` and the manager's own keys, the **history** row kept current in the thread-metadata store, and cheap **session metadata** (`snapshot_meta`) on the send path.
 - **`DeltaProjector`** (`crates/atlas-agent-delta`) turns thread events into the wire the rest of Atlas consumes.
 

crates/atlas-acp-thread/src/thread.rs

@@ -839,10 +839,11 @@ pub enum AcpThreadEvent {
 }
 
 /// The turn currently in flight, if any. Zed also holds the send task here; the
-/// task lives with the caller in this port, so only the identity remains.
+/// task lives with the caller in this port, so only the identity remains — and
+/// the identity is the point: it is what tells a turn reporting back late
+/// whether it is still the one running.
 #[derive(Debug)]
 struct RunningTurn {
-    #[allow(dead_code)]
     id: u32,
 }
 
@@ -1768,6 +1769,61 @@ impl AcpThread {
         self.emit(AcpThreadEvent::StatusChanged);
     }
 
+    /// Close `turn`, unless a later turn has taken over.
+    ///
+    /// The guarded counterpart to [`Self::end_turn`], for the caller that owns
+    /// one specific turn and may be reporting back long after it stopped being
+    /// the current one. `begin_turn` supersedes deliberately, so a turn
+    /// interrupted by a follow-up still has a `prompt` in flight; when it
+    /// finally returned, closing the thread's turn unconditionally closed the
+    /// turn that had superseded it, and the session left the generating state
+    /// while the live turn was still streaming (ATL-229).
+    ///
+    /// The guard is "no OTHER turn is running", not "this turn is running", and
+    /// the difference is the whole of the cancel path. `cancel()` clears
+    /// `running_turn` before the agent has answered, so by the time the prompt
+    /// returns `StopReason::Cancelled` there is no running turn at all — and a
+    /// guard that demanded one would swallow the only `Stopped` a cancelled
+    /// turn ever emits. Nothing downstream would then flush: no
+    /// `TurnFinished` on the wire, no analytics, no transcript write, and the
+    /// next turn's usage footer would carry the cancelled turn's tokens.
+    ///
+    /// So: no turn running means announce (this is the cancelled or
+    /// already-closed turn reporting its own end), a different turn running
+    /// means stay quiet.
+    ///
+    /// Answers whether it closed anything, so a caller can tell "my turn ended"
+    /// from "my turn had already been replaced".
+    pub fn end_turn_unless_superseded(&mut self, turn: u32, stop_reason: acp::StopReason) -> bool {
+        if self.is_superseded(turn) {
+            return false;
+        }
+        self.end_turn(stop_reason);
+        true
+    }
+
+    /// Mark the thread failed, unless a later turn has taken over.
+    ///
+    /// The guarded counterpart to [`Self::set_error`], with the same guard and
+    /// for the same reason as [`Self::end_turn_unless_superseded`]: a
+    /// superseded turn's failure is news about a turn that is already over, and
+    /// marking the live one as errored on the strength of it is a lie the user
+    /// acts on.
+    pub fn set_error_unless_superseded(&mut self, turn: u32) -> bool {
+        if self.is_superseded(turn) {
+            return false;
+        }
+        self.set_error();
+        true
+    }
+
+    /// Whether some turn other than `turn` is the one running.
+    fn is_superseded(&self, turn: u32) -> bool {
+        self.running_turn
+            .as_ref()
+            .is_some_and(|running| running.id != turn)
+    }
+
     pub fn emit_load_error(&mut self, error: LoadError) {
         self.had_error = true;
         self.running_turn = None;

crates/atlas-acp-thread/tests/thread.rs

@@ -515,6 +515,100 @@ async fn a_new_turn_clears_a_stale_call_left_by_a_failed_one() {
     assert_eq!(status_of(&thread, "t1"), "Canceled");
 }
 
+// ----------------------------------------------------------------- turn ids
+
+/// Regression, ATL-229. `begin_turn` returns an id precisely so a turn that
+/// reports back late can tell whether it is still the one running. Closing the
+/// thread's turn unconditionally meant the *cancelled* turn closed the *live*
+/// one, and the UI dropped to idle mid-stream.
+#[tokio::test]
+async fn a_superseded_turn_cannot_close_the_turn_that_replaced_it() {
+    let (mut thread, _events, _conn) = new_thread();
+
+    let first = thread.begin_turn();
+    let second = thread.begin_turn();
+    assert_ne!(first, second, "each turn gets its own id");
+
+    assert!(
+        !thread.end_turn_unless_superseded(first, acp::StopReason::Cancelled),
+        "the superseded turn closes nothing"
+    );
+    assert!(thread.is_generating(), "the live turn is still running");
+
+    assert!(
+        thread.end_turn_unless_superseded(second, acp::StopReason::EndTurn),
+        "and the live turn closes its own"
+    );
+    assert!(!thread.is_generating());
+}
+
+/// The same guard for the failing half: a superseded turn's error is news about
+/// a turn that is already over.
+#[tokio::test]
+async fn a_superseded_turns_error_does_not_mark_the_live_turn() {
+    let (mut thread, _events, _conn) = new_thread();
+
+    let first = thread.begin_turn();
+    thread.begin_turn();
+
+    assert!(!thread.set_error_unless_superseded(first));
+    assert!(thread.is_generating(), "the live turn survives it");
+    assert!(!thread.had_error(), "and is not marked as having failed");
+}
+
+/// The guard is "no other turn is running", not "this turn is running", and
+/// this is why. `cancel()` clears `running_turn` before the agent answers, so
+/// the prompt returns `Cancelled` into a thread with no turn open. A stricter
+/// guard would swallow the only `Stopped` a cancelled turn ever emits — and
+/// `Stopped` is what produces `TurnFinished` on the wire, which is what flushes
+/// analytics, the transcript and the turn's token usage.
+#[tokio::test]
+async fn a_cancelled_turn_still_announces_its_own_stop() {
+    let (mut thread, mut events, _conn) = new_thread();
+
+    let turn = thread.begin_turn();
+    thread.cancel();
+    assert!(!thread.is_generating(), "cancel already cleared the turn");
+    while events.try_recv().is_ok() {}
+
+    // What the agent sends back once it has acknowledged the cancel.
+    assert!(
+        thread.end_turn_unless_superseded(turn, acp::StopReason::Cancelled),
+        "the cancelled turn closes itself"
+    );
+
+    let mut stopped = None;
+    while let Ok(event) = events.try_recv() {
+        if let AcpThreadEvent::Stopped(reason) = event {
+            stopped = Some(reason);
+        }
+    }
+    assert_eq!(
+        stopped,
+        Some(acp::StopReason::Cancelled),
+        "the stop is announced, with the reason the agent gave"
+    );
+}
+
+/// The unguarded `end_turn` stays as it is: `atlas-agent-delta` and this
+/// crate's own callers use it as "announce Stopped" with no turn open, which
+/// ATL-218 finding 4 settled deliberately. The guarded pair is an addition, not
+/// a replacement.
+#[tokio::test]
+async fn end_turn_still_announces_a_stop_with_no_turn_open() {
+    let (mut thread, mut events, _conn) = new_thread();
+
+    thread.end_turn(acp::StopReason::EndTurn);
+
+    let mut stopped = false;
+    while let Ok(event) = events.try_recv() {
+        if matches!(event, AcpThreadEvent::Stopped(_)) {
+            stopped = true;
+        }
+    }
+    assert!(stopped, "the stop is announced whether or not a turn was open");
+}
+
 /// The counterpart guard: resolving entries unconditionally must not start
 /// sending `session/cancel` from an idle thread. Complements
 /// `cancelling_an_idle_thread_does_not_notify_the_agent` by putting a

crates/atlas-agent-manager/src/lib.rs

@@ -13,26 +13,43 @@
 //!
 //! - **One connect attempt per agent.** A second `request_connection` while the
 //!   first is still connecting joins it instead of starting a second process;
-//!   that is what the shared connect future is for.
+//!   that is what the shared connect future is for. The check and the insert
+//!   happen under one `entries` guard, because they are one decision — split,
+//!   two concurrent callers each started a process (ATL-226).
 //! - **A failed connection does not stick.** The entry is set to `Error` (so a
 //!   waiter sees why) *and* removed from the table, so the next request
 //!   reconnects rather than replaying the old failure forever.
 //! - **A version bump drops the connection.** When the store reports the agent
 //!   moved forward, the entry goes; the running process is on the old binary and
 //!   the next request starts the new one.
 //!
+//! Every path that evicts an entry also forgets that agent's sessions and stops
+//! any connect still in flight. A session pins the connection, so one left
+//! behind keeps a child process alive that nothing can reach (ATL-227); and an
+//! attempt nobody stops finishes its download, spawns its process and completes
+//! its handshake for an agent the user already killed (ATL-228).
+//!
 //! # Not ported
 //!
 //! GPUI. Zed's entries are `Entity<AgentConnectionEntry>` compared by identity,
 //! its connect is a `Task`, and its notifications are `cx.emit`. Here those are
 //! `Arc<Mutex<_>>` compared with `Arc::ptr_eq`, a `Shared` future, and a
-//! broadcast channel. The mechanism is unchanged; only the runtime is.
+//! broadcast channel.
+//!
+//! That substitution is not free, and this file used to claim it was: "the
+//! mechanism is unchanged; only the runtime is". It is the other way round.
+//! Zed's store is `Rc`-based and cannot leave the GPUI main thread, so its
+//! check-then-insert is one uninterruptible borrow and its correctness came
+//! from the runtime it ran on. Ported onto multi-threaded tokio with the same
+//! shape, the guarantee stopped holding. Anything else moved across from
+//! Zed's GPUI-side code deserves the same question: what was the original
+//! leaning on that did not come with it?
 
 pub mod catalog;
 pub mod manager;
 
 pub use catalog::AgentCatalog;
 pub use manager::{
     Agent, AgentConnectedState, AgentConnectionEntry, AgentConnectionStatus, AgentManager,
-    AgentManagerEvent, ResumeMode, ResumedSession, SessionHandle,
+    AgentManagerEvent, ConnectHandle, ResumeMode, ResumedSession, SessionHandle,
 };