From bb475856a09d6f188473dd8989ea1f66f3d46333 Mon Sep 17 00:00:00 2001 From: kosmos Date: Wed, 11 Nov 2026 18:21:43 +0000 Subject: [PATCH] fix(claw): a job dispatch drains its inbox, so its key goes idle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A finished coding job's key never settles. The handler takes its work from the job table and never polls the inbox, so the item that opened the dispatch stays there; `release` reads a non-empty inbox as work that arrived mid-turn, `end_lease` sends the key to re-acquire with a fresh debounce window, and `admit` hands out a lease when it lapses. The re-dispatched handler finds the row done, answers `Handled`, and the key re-acquires again: one lease and two record writes per debounce window, per finished job, for the rest of the process's life. Seen on a live daemon: `job:rhema-claw#1..#3` sat in `reacquire` with `inbox_depth` 1, revisions climbing two every ~40s, hours after the jobs were done — and every surface that reads the busy rule (`inbox_depth == 0 && state != reacquire`) called them busy. Every other handler polls: `interactive.rs` folds the batch into the turn's input, `handler.rs` drains the forge news. This one has no news to fold — a job key carries the row's own delivery — so it drains and drops. The poll's other half matters too: it renews the lease, which a coding turn running for minutes needs. The keys already spinning on a running daemon settle by themselves on their next dispatch. Test: a queued job runs to done through the source and the driver, and the key it leaves behind is idle with an empty inbox and nothing due on the next admit pass — the state a leftover item cannot produce. --- crates/rhema-claw/src/jobs.rs | 70 +++++++++++++++++++++++++++++++---- 1 file changed, 63 insertions(+), 7 deletions(-) diff --git a/crates/rhema-claw/src/jobs.rs b/crates/rhema-claw/src/jobs.rs index e275f23..4b0709d 100644 --- a/crates/rhema-claw/src/jobs.rs +++ b/crates/rhema-claw/src/jobs.rs @@ -167,6 +167,20 @@ impl CodingJobHandler { /// lands in the job row before it lands in the outcome, so the /// model's `status` read and the user's notification agree. async fn dispatch(&self, ctx: &mut SessionContext) -> Result { + // Poll to live, as every handler does — and here the poll is also + // what closes the dispatch. A job key carries exactly one kind of + // item, the row's own delivery, and those items are drained and + // dropped because this handler takes its work from the store, not + // from the inbox. Leaving them behind is not harmless: `release` + // reads a non-empty inbox as work that arrived mid-turn, sends the + // key to `reacquire`, and `admit` then hands out a fresh lease + // every debounce window for the rest of the process's life — for a + // job that is already done. The renewing half matters too: a + // coding turn runs for minutes, and the lease it started on has to + // hold for all of them. + ctx.poll_inbox() + .await + .map_err(|err| format!("inbox poll for {} failed: {err}", ctx.key))?; let (instance_name, job_id) = parse_job_key(&ctx.key) .ok_or_else(|| format!("resource key {} is not a job key", ctx.key))?; let job = self @@ -333,7 +347,7 @@ mod tests { use crate::backends::Backend; use rhema_autonomous_agent::Driver; use rhema_core::{Message as CoreMessage, MockProvider, ModelOutput, Role, StopReason, Usage}; - use rhema_rsd::{DispatcherConfig, SqliteStore, SystemClock}; + use rhema_rsd::{DispatcherConfig, KeyState, SqliteStore, SystemClock}; /// A registry whose only backend is the scripted mock. fn mock_handler( @@ -380,7 +394,7 @@ mod tests { // The whole path: source detects the queued row, the driver // runs the handler over its own dispatcher. - let (source, driver) = harness(store.clone(), handler); + let (source, driver, _dispatcher) = harness(store.clone(), handler); assert_eq!(source.detect().await.unwrap(), 1); let outcomes = driver.run_once().await.unwrap(); @@ -418,7 +432,7 @@ mod tests { second.to_string().contains("job in flight"), "one live job per instance: {second}" ); - let (source, driver) = harness(store.clone(), handler); + let (source, driver, _dispatcher) = harness(store.clone(), handler); assert_eq!(source.detect().await.unwrap(), 1); driver.run_once().await.unwrap(); @@ -452,7 +466,7 @@ mod tests { store.clone(), notifications.clone(), ); - let (source, driver) = harness(store.clone(), handler); + let (source, driver, _dispatcher) = harness(store.clone(), handler); assert_eq!(source.detect().await.unwrap(), 1); let outcomes = driver.run_once().await.unwrap(); @@ -470,7 +484,10 @@ mod tests { /// The source and driver over one in-memory dispatcher, the shape /// the daemon wires and these tests drive. - fn harness(store: Arc, handler: CodingJobHandler) -> (JobsSource, Driver) { + fn harness( + store: Arc, + handler: CodingJobHandler, + ) -> (JobsSource, Driver, Arc) { let rsd_store = Arc::new(SqliteStore::open(tempdir().join("rsd.db")).unwrap()); // A zero debounce: the queued row is admitted on this pass, // the shape the daemon's jobs dispatcher runs under. @@ -483,9 +500,48 @@ mod tests { rsd_store, )); let source = JobsSource::new(&dispatcher, store); - let driver = Driver::new(dispatcher, Arc::new(handler)) + let driver = Driver::new(dispatcher.clone(), Arc::new(handler)) .with_holder("test-jobs") .owning(Arc::new(|key: &ResourceKey| is_job_key(key.as_str()))); - (source, driver) + (source, driver, dispatcher) + } + + /// A finished job's key settles back to idle. + /// + /// The item that opened the dispatch is drained by the handler, so + /// `release` finds an empty inbox and ends the key. Left behind, that + /// item reads as work that arrived mid-turn: the key enters + /// `reacquire` with a fresh debounce window, `admit` hands out a + /// lease when it lapses, the re-dispatched handler finds the row done + /// and releases, and the key re-acquires again — a finished job's key + /// holding leases and writing records for the rest of the process's + /// life. + #[tokio::test] + async fn a_finished_job_leaves_its_key_idle() { + let workspace = tempdir(); + let store = Arc::new(CodingJobStore::in_memory().unwrap()); + store + .add_instance("rhema", workspace.to_str().unwrap()) + .unwrap(); + let job = store.enqueue("rhema", "fix the lint").unwrap(); + let notifications = Arc::new(NotificationStore::in_memory().unwrap()); + let handler = mock_handler(vec![reply("done")], store.clone(), notifications); + let (source, driver, dispatcher) = harness(store, handler); + + assert_eq!(source.detect().await.unwrap(), 1); + assert_eq!(driver.run_once().await.unwrap().len(), 1); + + let key = ResourceKey::from(job_key("rhema", job.id)); + let status = dispatcher.status(&key).unwrap(); + assert_eq!( + status.state, + KeyState::Idle, + "a done job's key is idle, not waiting to re-acquire" + ); + assert_eq!(status.inbox_depth, 0, "the dispatching item was drained"); + + // Nothing is due, so the next pass dispatches nothing: the + // re-acquire spin a leftover item would drive, and nothing else. + assert!(driver.run_once().await.unwrap().is_empty()); } }