Check "exited" against the process record before believing it

A session adopted at a backend start keeps the transcript's last status,
so one whose process had been reported gone and was then found again
read as `exited` while its CLI was running. `exited` is the word that
draws the phone's Start button and lets `start_session` build a driver,
so Start was accepted every time it was pressed -- and since starting
replaces the driver without retiring the old one, each press left
another reader on the same process. Every line the CLI wrote was then
translated once per reader: three presses put three interleaved copies
of one reply on screen, which is what it was reported as.

So `exited` is now checked against `session::process`, the one authority
on whether a process exists, in `launch` and again in `start_session`. A
record that is not known to be dead makes it false, and what replaces it
is `unknown` -- there is a process, and nothing here has heard from it,
which is the answer `status_of_unlaunched` already gave to the same
question. The correction goes out through the sink rather than into the
manager's view alone, or the list and the session screen would disagree
about it in the way this same button did a commit ago.

A driver that `start_session` replaces now gets `Driver::detach`, which
already existed for the backend going away and is the whole of what a
driver whose process has exited is owed.

On the phone the process button is disabled while its own request is in
flight, so a second press cannot be decided against a status the first
has not changed yet. That is a courtesy rather than the fix; the server
refuses it either way, because a phone that has lost the stream cannot
be relied on to know.

Verified against a stand-in CLI, with the state forced by hand: before,
three Starts returned 204 and left four readers on one process and the
status still `exited`; after, the session reports `unknown` on both
surfaces and all three are refused. Then driven on the emulator --
Stop, Start, Stop, Start alternated correctly with one process at a
time, and the list, the transcript and the record all agree.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

server/src/session/mod.rs

@@ -1132,16 +1132,30 @@ impl SessionManager {
             .with_context(|| format!("no session {id}"))?
             .clone();
         let existing = inner.live.get(id).cloned();
+        let dir = self.data_dir.join(id);
         let status = match &existing {
-            Some(session) => *session.shared.status.lock().unwrap(),
-            None => status_of_unlaunched(&self.data_dir.join(id)),
+            Some(session) => {
+                let last = *session.shared.status.lock().unwrap();
+                let now = corrected(last, &dir);
+                if now != last {
+                    // Published, not merely acted on. The phone is drawing a
+                    // Start button on the strength of the word this has just
+                    // disproved, and it learns what a session is doing from
+                    // the stream like everything else -- so a correction
+                    // nobody sends leaves that button there to be pressed
+                    // again, and again. Through the sink, which keeps the
+                    // pump the only writer of the status.
+                    let _ = session.sink.send(Event::Status { state: now });
+                }
+                now
+            }
+            None => status_of_unlaunched(&dir),
         };
         match status {
             SessionStatus::Exited => {}
-            SessionStatus::Unknown => bail!(
-                "this machine won't say whether this session's process is still running, so \
-                 nothing was started"
-            ),
+            SessionStatus::Unknown => {
+                bail!("there is still a process recorded for this session, so nothing was started")
+            }
             _ => bail!("this session is already running"),
         }
         // Fresh from the config, like every other launch: a model or a
@@ -1150,6 +1164,12 @@ impl SessionManager {
         let (setup, provider) = resolve(&inner.config, &meta)?;
         match existing {
             Some(session) => {
+                // The driver being replaced is still reading this session's
+                // output, and replacing the value it lives in does not end
+                // the tasks that do it. Its process has exited -- that is
+                // how this line was reached -- so there is nothing left to
+                // preserve and `detach` is the whole of what it is owed.
+                session.driver().detach();
                 *session.driver.lock().unwrap() = make_driver(
                     &meta,
                     &setup,
@@ -1226,6 +1246,38 @@ impl SessionManager {
 /// reports `Unknown` too: this server is not driving it, so it genuinely
 /// does not know what it is doing -- and that is worth a word that means
 /// "wait", not one that means "act".
+/// The last word about a session, with the one status that cannot be taken
+/// on trust checked against the only authority on it.
+///
+/// `Exited` is not just a description: it is the word that offers a phone a
+/// Start button and lets [`SessionManager::start_session`] build a second
+/// CLI against a conversation. So before it is believed it is checked
+/// against the process record, and a record that is not known to be dead
+/// makes it false. What replaces it is `Unknown` -- there is a process, and
+/// nothing here has heard from it -- which is the same answer
+/// [`status_of_unlaunched`] gives to the same question.
+///
+/// Every other status is left exactly as it was. Those are the pump's,
+/// written from what the process itself said, and none of them authorises
+/// starting anything.
+///
+/// This was reachable and did happen: a session adopted at server start
+/// keeps the transcript's last word, so one whose process was reported gone
+/// and then found again read as `exited` while it was running. Start was
+/// accepted every time it was pressed, each press attaching another reader
+/// to the one process, and every line it wrote was then translated once per
+/// reader -- three presses put three interleaved copies of one reply on
+/// screen.
+fn corrected(status: SessionStatus, session_dir: &Path) -> SessionStatus {
+    if status != SessionStatus::Exited {
+        return status;
+    }
+    match process::recorded(session_dir) {
+        Some((_, process::Liveness::Alive | process::Liveness::Unknown)) => SessionStatus::Unknown,
+        Some((_, process::Liveness::Dead)) | None => SessionStatus::Exited,
+    }
+}
+
 fn status_of_unlaunched(session_dir: &Path) -> SessionStatus {
     match process::recorded(session_dir) {
         // Nothing was ever recorded: an echo session, or one whose
@@ -1422,6 +1474,7 @@ fn launch(
     wg_app_link::private::create_dir(&dir)?;
     let transcript_path = dir.join("transcript.jsonl");
     let mut transcript = Transcript::open(&transcript_path)?;
+    let last_status = transcript.last_status().unwrap_or(SessionStatus::Idle);
     // Before the driver starts, so the token is there when it looks and
     // the history is already in the transcript a phone will read.
     if let Some(seed) = seed {
@@ -1439,8 +1492,11 @@ fn launch(
         // What it was last known to be doing, not an assumption. A driver
         // that has something to say corrects this within its first poll;
         // one adopting a process that has been quiet says nothing, and
-        // this is then the only true answer available.
-        status: Mutex::new(transcript.last_status().unwrap_or(SessionStatus::Idle)),
+        // this is then the only true answer available. Where it is *not*
+        // the true answer, the correction goes through the sink below
+        // rather than being written here, so that it reaches the
+        // transcript too -- see there.
+        status: Mutex::new(last_status),
         title: Mutex::new(meta.title.clone()),
         // What the transcript last recorded, not the clock: this server has
         // just been told nothing, and `now()` claimed every relaunched
@@ -1494,6 +1550,22 @@ fn launch(
         );
     }
 
+    // Before the driver, because the driver may start a process and say so,
+    // and that has to be the later word of the two.
+    //
+    // Sent rather than written into `shared`: the session screen replays the
+    // transcript and the session list reads `shared`, so a correction made
+    // in only one of them is the two of them describing one session
+    // differently -- which is how the phone came to show a Start button on a
+    // running session in the first place. One event, and the pump puts it in
+    // both.
+    let corrected_status = corrected(last_status, &dir);
+    if corrected_status != last_status {
+        let _ = sink.send(Event::Status {
+            state: corrected_status,
+        });
+    }
+
     let driver = Arc::new(Mutex::new(make_driver(
         &meta,
         setup,
@@ -2493,6 +2565,76 @@ mod tests {
         ));
     }
 
+    /// The status is a claim about a process, and the process record is
+    /// what settles it.
+    ///
+    /// Without this the phone offered Start on a session whose CLI was
+    /// running, and taking it up attached a second reader to that one
+    /// process rather than failing -- so the session went on saying
+    /// `exited`, the button stayed, and each further press added another
+    /// reader. On screen that was one reply written as many times as the
+    /// button had been pressed, interleaved word by word.
+    #[tokio::test]
+    async fn a_stale_exited_does_not_start_anything_while_a_process_is_recorded() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let config_path = dir.path().join("config.ron");
+        let data_dir = dir.path().join("sessions");
+        seed_echo_only(&config_path);
+        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
+            .expect("manager");
+        let info = manager.spawn_session(echo_spec()).expect("spawn");
+        let session = manager.session(&info.id).expect("live session");
+        let mut rx = session.subscribe();
+
+        // A live process for this session: this test's own, which is the
+        // one process certain to still be there when the guard looks.
+        let record = process::Record::of(
+            std::process::id(),
+            process::Detail::Stdio { stdout_read: 0 },
+        )
+        .expect("record this process");
+        process::write(&data_dir.join(&info.id), &record);
+        let _ = session.sink.send(Event::Status {
+            state: SessionStatus::Exited,
+        });
+        collect_until(&mut rx, |event| {
+            matches!(
+                event,
+                Event::Status {
+                    state: SessionStatus::Exited
+                }
+            )
+        })
+        .await;
+
+        let refused = manager
+            .start_session(&info.id)
+            .expect_err("a process is recorded");
+        assert!(
+            refused.to_string().contains("still a process recorded"),
+            "said: {refused:#}"
+        );
+        // And the word that was wrong is taken back, on the stream and in
+        // the transcript -- otherwise the button that asked for this is
+        // still there, still saying Start.
+        collect_until(&mut rx, |event| {
+            matches!(
+                event,
+                Event::Status {
+                    state: SessionStatus::Unknown
+                }
+            )
+        })
+        .await;
+        assert_eq!(manager.sessions()[0].status, SessionStatus::Unknown);
+        assert_eq!(
+            Transcript::open(&data_dir.join(&info.id).join("transcript.jsonl"))
+                .expect("reopen transcript")
+                .last_status(),
+            Some(SessionStatus::Unknown),
+        );
+    }
+
     #[tokio::test]
     async fn a_restart_relaunches_sessions_and_continues_the_numbering() {
         let dir = tempfile::tempdir().expect("tempdir");