From 160b0bded267ddac15c979a3dbad998ea802ea36 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Oct 2026 03:48:12 +0000 Subject: [PATCH] Two daemon test flakes: a stop that waited on background commands, and a test lock that held the service loop A daemon that stopped with a background command still running waited up to 5 s for the publication stream: the command's result task kept a store handle open until the command ended or the stream timed out. The tests' fixture times commands out at 5 s, so the stop took about as long as `Client.close` allows, failed it now and then, and recorded the command as timed out. The registry now ends background commands as the daemon stops, as it ends terminals: each group is killed and its handle records process_lost at once, which the next start would have said. The fatal-storage test holds SQLite's write lock while it sends its admissions. The service looks for ready turns as each turn ends, on its own loop and through the store; a look still to come when the lock was taken held that loop, so the admissions were read only after the failure. The test now waits until the service has looked, by its store counters, before taking the lock. Starving the daemon of CPU reproduced the missing replies in 2 of 8 runs before and none of 12 after. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013jkdGEgwvrmgA2w4Bq4gqN --- docs/RUST_PROTOTYPE.md | 3 ++- src/server/mod.rs | 4 +++- src/tools.rs | 23 ++++++++++++++++++++--- tests/test_wait.py | 33 +++++++++++++++++++++++++++++++++ 4 files changed, 58 insertions(+), 5 deletions(-) diff --git a/docs/RUST_PROTOTYPE.md b/docs/RUST_PROTOTYPE.md index 234f454ce..ab5113a6f 100644 --- a/docs/RUST_PROTOTYPE.md +++ b/docs/RUST_PROTOTYPE.md @@ -251,7 +251,8 @@ Parked turns survive a daemon restart: they are re-registered at startup, peer handles resolve from durable state (a peer interrupted by the restart reports `interrupted`), and commands that were still running resolve to `process_lost`: their supervision ended with the daemon, whether or not the -OS process did. Interrupting a parked +OS process did. A daemon that stops cleanly does not wait for them: it kills +each one's process group and records `process_lost` itself. Interrupting a parked turn ends it as `interrupted`: the wait and any planned calls behind it get a `cancelled` tool result in the same transaction, so the conversation stays valid and the bot is not left uncertain. A bot whose turn is parked reports diff --git a/src/server/mod.rs b/src/server/mod.rs index 12d223fac..0dc7312cf 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1596,8 +1596,10 @@ pub async fn run(config: Configuration) -> Result<()> { } service.replays.abort_all(); while service.replays.join_next().await.is_some() {} - // Terminals end with the daemon, and record how on their handles. + // Terminals and background commands end with the daemon, and record + // how on their handles, so no result task keeps the store open below. service.registry.end_terminals(); + service.registry.end_background(); // Every completion is committed. Allow publication to drain before // releasing waiters. Background result tasks may still own Store clones, // keeping the stream open even after the service is dropped. diff --git a/src/tools.rs b/src/tools.rs index e271734e2..52663b4e8 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -18,7 +18,7 @@ use std::{ }; use tokio::{ io::{AsyncRead, AsyncReadExt}, - sync::Semaphore, + sync::{Semaphore, watch}, }; mod fetch; @@ -412,6 +412,9 @@ pub struct Registry { /// Running terminals, by proc handle, each its bot's own. #[cfg(unix)] terminals: terminal::Terminals, + /// Set once as the daemon stops: background commands still running + /// end then rather than outlive it. + stopping: Arc>, credentials: Credentials, environment: Arc>, /// What the daemon keeps that a sandbox leaves alone or lets through. @@ -894,6 +897,7 @@ impl Registry { detached_budget: DEFAULT_DETACHED_BUDGET, #[cfg(unix)] terminals: Arc::default(), + stopping: Arc::new(watch::Sender::new(false)), credentials: Credentials::default(), environment: Arc::new(Vec::new()), daemon: Arc::default(), @@ -1714,6 +1718,7 @@ impl Registry { ) { let registry = self.clone(); let mut queued = queued; + let mut stopping = self.stopping.subscribe(); tokio::spawn(async move { let result = async { let slot = match queued.permit.take() { @@ -1733,11 +1738,23 @@ impl Registry { ) .await?; Ok(registry.shell_outcome(stdout, stderr, exit, elapsed)) + }; + // A command still running as the daemon stops is dropped, which + // kills its group; its handle then reports the process lost, as + // the next start would. + tokio::select! { + result = result => { + let _ = done.send(result); + } + _ = stopping.wait_for(|stopping| *stopping) => {} } - .await; - let _ = done.send(result); }); } + + /// End every background command still running, as the daemon stops. + pub fn end_background(&self) { + self.stopping.send_replace(true); + } } /// Number lines from a 1-based offset within the page budget. A single line diff --git a/tests/test_wait.py b/tests/test_wait.py index ebeb5ad78..443c5a62e 100644 --- a/tests/test_wait.py +++ b/tests/test_wait.py @@ -108,6 +108,24 @@ def background(request, command): self.assertLess(process.memory_info().rss - baseline, 8 * 1024 * 1024, phase) self.assertEqual(client.request('wait', handles=[pending], timeout_ms=0)['result']['pending'], [pending]) + def test_a_clean_stop_ends_a_running_background_command_and_records_it_lost(self): + client = self.client('echo,shell,wait') + client.request('create', bot='Bob', workspace=str(self.path)) + turn = client.request('submit', bot='Bob', request_id='bg', + prompt='bg:while [ ! -f finish ]; do sleep .01; done')['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + handle = self.tool_output(client, 'Bob', 'bg-1')['handle'] + # The stop ends the command and records its end itself, instead of + # leaving the row running for the next start to find. + client.close() + with sqlite3.connect(self.path / 'state.sqlite') as db: + rows = db.execute('SELECT status,result FROM processes').fetchall() + self.assertEqual([(status, json.loads(result)['error']) for status, result in rows], + [('finished', 'process_lost')]) + again = self.client('echo,shell,wait') + result = again.request('wait', handles=[handle])['result']['results'][handle] + self.assertEqual(result['error'], 'process_lost') + def test_old_turn_outcomes_use_indexed_event_lookups(self): client = self.client() client.request('create', bot='Bob', workspace=str(self.path)) @@ -759,9 +777,24 @@ def test_a_fatal_storage_failure_still_answers_admissions_queued_before_it(self) client = self.client('echo,shell,wait') for bot in ('Bob', 'Carol'): client.request('create', bot=bot, workspace=str(self.path)) + # The service looks for ready turns once as it starts and again as + # each turn ends, on its own loop and through the store. A look still + # to come when the lock below is taken would hold that loop, so it + # would read the admissions only after the failure: wait for each. + def looks(at_least): + deadline = time.monotonic() + 10 + while True: + operations = client.request('stats')['result']['store']['operations'] + count = operations.get('next_ready', {}).get('count', 0) + if count >= at_least: + return count + self.assertLess(time.monotonic(), deadline) + time.sleep(.01) + before = looks(1) started = client.request('submit', bot='Bob', request_id='bg', prompt="bg:while [ ! -f release ]; do sleep .01; done; printf done > marker")['result']['turn'] self.assertEqual(client.finished(started)['data']['status'], 'completed') + looks(before + 1) with sqlite3.connect(database) as db: db.execute("CREATE TRIGGER reject_completion BEFORE UPDATE OF result ON processes BEGIN " "SELECT RAISE(ABORT, 'synthetic write failure'); END")