diff --git a/AGENTS.md b/AGENTS.md index 7c5782e..96eccf9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -302,6 +302,17 @@ day: launch has just started a process reports `idle`, because `exited` is the word that refuses every command and offers a phone the chance to start a second CLI on a live conversation. +- **`exited` is never taken on trust; it is checked against the process + record** (`corrected` in `session/mod.rs`). It is the one status that draws + the phone's Start button and lets `start_session` build a driver, so a + record that is not known to be dead makes it false and the session reports + `unknown` instead. Without that, a session adopted at a backend start kept + the transcript's `exited` while its CLI was running, Start was accepted + every press, and each press left another reader on the same process — + which reads on screen as one reply written several times, interleaved + (`GotGotGot it — it — it —`), not as anything to do with a button. + A driver that `start_session` replaces gets `Driver::detach` for the same + reason: swapping the `Arc` does not end the tasks the old one is running. - Remote sessions are adopted too. The pid recorded for one is the **`ssh` client's**, on this machine — that is the process the backend owns, and it lives as long as the remote command does. (This said "local only" until diff --git a/PLAN.md b/PLAN.md index d2be19c..d069af7 100644 --- a/PLAN.md +++ b/PLAN.md @@ -351,6 +351,36 @@ better answer until its output says otherwise. Coming from the driver also orders it against the exit `follow` reports, which a status written from the manager could not be. +**`Exited` is a claim about a process, and the record is what settles it.** +Adopting saying nothing left one word standing that a live process +contradicts. A session whose process was reported gone and then found again +at the next backend start kept `Exited` from the transcript — and `Exited` is +the word that draws a Start button. Start was then accepted every time it was +pressed, and since starting replaces the driver, each press attached *another* +reader to the one process: every line the CLI wrote was translated once per +reader, so three presses put three interleaved copies of one reply on screen. +Two rules come out of it, and neither is optional: + +- **`Exited` is checked against `session::process` before it is believed** — + `corrected`, called 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. 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. The + correction goes out through the sink for the reason above: written into the + manager's view alone it would be the list and the screen disagreeing again. +- **A driver that is replaced is detached.** Swapping the `Arc` does not end + the tasks the old one is running. `Driver::detach` is what does — it already + existed for the backend going away — and it is the whole of what a driver + whose process has exited is owed. + +The phone's half is that the process button is disabled while its own request +is in flight, so a second press cannot be decided against a status the first +one has not changed yet. That is a courtesy rather than the fix: the server +refuses the second request either way, because a phone that has lost the +stream cannot be relied on to know. + On the phone this is one button in the composer, left of Send, whose mark and colour say what pressing it would do now: an orange pause while a turn is running (interrupt — the process stays), a red stop when it is not (end the diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index 567a0cd..9bb7500 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -555,6 +555,12 @@ fun SessionScreen( var compactingFor by remember { mutableStateOf(null) } var streamError by remember { mutableStateOf(null) } var actionError by remember { mutableStateOf(null) } + // Whether the composer's process button has a request out. What it does next is decided from + // the session's status, and the status only changes once the server has answered and the + // stream has carried it back -- so two presses in that gap are two requests, both decided + // against the state before either of them. The server refuses the second one, but a control + // that can be pressed while its own last press is still in flight is asking to be. + var processInFlight by remember { mutableStateOf(false) } val context = LocalContext.current // Seeded from what was left in the box last time and written back on every keystroke, so // leaving the screen -- or the system reclaiming the app -- does not throw away a half-typed @@ -997,7 +1003,7 @@ fun SessionScreen( } } - fun act(onFailure: () -> Unit = {}, action: () -> Unit) { + fun act(onFailure: () -> Unit = {}, onDone: () -> Unit = {}, action: () -> Unit) { scope.launch { try { withContext(Dispatchers.IO) { action() } @@ -1005,6 +1011,11 @@ fun SessionScreen( } catch (e: ApiException) { actionError = e.message onFailure() + } finally { + // Whatever happened, including the failure above: a caller that re-enables a + // control here must get it back on the path where the request was refused too, + // or the refusal is what disables the control permanently. + onDone() } } } @@ -1503,7 +1514,13 @@ fun SessionScreen( else -> ProcessAction.Stop } Button( - onClick = { act { process.perform(settings, summary.id) } }, + onClick = { + processInFlight = true + act(onDone = { processInFlight = false }) { + process.perform(settings, summary.id) + } + }, + enabled = !processInFlight, colors = actionButtonColors(process.colour()), ) { Glyph( diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index f29f5eb..f4fe012 100644 --- a/server/src/session/mod.rs +++ b/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");