fix: fence cleanup on tracked run completion

crates/worker/src/controller.rs

@@ -300,6 +300,7 @@ async fn finish_controller_run<C, St>(
     runtime_dir: &RuntimeDir,
     working_event_tx: &broadcast::Sender<Event>,
     new_status: WorkerStatus,
+    finished_submission_request_id: Option<String>,
 ) where
     C: LlmClient + Clone + 'static,
     St: Store + WorkerMetadataStore + Clone + 'static,
@@ -320,7 +321,20 @@ async fn finish_controller_run<C, St>(
         Some(worker.total_tokens().tokens),
         protocol::ContextTokenSource::Estimated,
     );
-    set_controller_status(shared_state, runtime_dir, working_event_tx, new_status).await;
+    if let Some(submission_request_id) = finished_submission_request_id {
+        let state = match new_status {
+            WorkerStatus::Idle | WorkerStatus::Stopped => WorkerState::Idle,
+            WorkerStatus::Running => {
+                WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running))
+            }
+            WorkerStatus::Paused => WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Paused)),
+        };
+        let snapshot = shared_state.finish_submission_run(state, submission_request_id);
+        let _ = runtime_dir.write_status(shared_state).await;
+        let _ = working_event_tx.send(Event::WorkerState { snapshot });
+    } else {
+        set_controller_status(shared_state, runtime_dir, working_event_tx, new_status).await;
+    }
 }
 
 /// Pending turn launch staged by an event handler for the next outer-loop
@@ -2023,6 +2037,10 @@ async fn controller_loop<C, St>(
             // interrupted/error turn from being carried into the next snapshot.
             worker.clear_in_flight_events();
             let parent_originated = run.is_parent_originated();
+            let finished_submission_request_id = match &run {
+                PendingRun::Submit(submission) => Some(submission.submission_request_id.clone()),
+                PendingRun::RunForNotification { .. } | PendingRun::Resume => None,
+            };
             let user_input_submit = matches!(&run, PendingRun::Submit(_));
             if !user_input_submit {
                 set_controller_status(
@@ -2121,12 +2139,16 @@ async fn controller_loop<C, St>(
                     }
                 }
             }
+            let finished_submission_request_id = (new_status != WorkerStatus::Paused)
+                .then_some(finished_submission_request_id)
+                .flatten();
             finish_controller_run(
                 &mut worker,
                 &shared_state,
                 &runtime_dir,
                 &working_event_tx,
                 new_status,
+                finished_submission_request_id,
             )
             .await;
             if shutdown {

crates/worker/src/internal_worker.rs

@@ -335,6 +335,7 @@ fn send_internal_worker_state(
     let _ = event_tx.send(Event::WorkerState {
         snapshot: protocol::WorkerStateSnapshot {
             last_command_id: 0,
+            last_finished_submission_request_id: None,
             state,
         },
     });

crates/worker/src/shared_state.rs

@@ -117,6 +117,23 @@ impl WorkerSharedState {
         snapshot.clone()
     }
 
+    /// Publish the post-Run state and the exact Submit request whose controller
+    /// invocation has returned. Updating both in one snapshot prevents observers
+    /// from treating an earlier Idle state as evidence for a later Run.
+    pub(crate) fn finish_submission_run(
+        &self,
+        state: WorkerState,
+        submission_request_id: String,
+    ) -> WorkerStateSnapshot {
+        let mut snapshot = self
+            .state
+            .write()
+            .expect("worker state lock poisoned; refusing an inferred fallback state");
+        snapshot.state = state;
+        snapshot.last_finished_submission_request_id = Some(submission_request_id);
+        snapshot.clone()
+    }
+
     pub(crate) fn admit_command(
         &self,
         envelope: WorkerCommandEnvelope,
@@ -305,6 +322,14 @@ mod tests {
         assert_eq!(snapshot.state, paused);
         assert_eq!(snapshot.last_command_id, 0);
         assert_eq!(state.catalog_status(), WorkerStatus::Paused);
+
+        let finished = state.finish_submission_run(WorkerState::Idle, "request-1".to_string());
+        assert_eq!(finished.state, WorkerState::Idle);
+        assert_eq!(
+            finished.last_finished_submission_request_id.as_deref(),
+            Some("request-1")
+        );
+        assert_eq!(finished.last_command_id, 0);
     }
 
     #[test]
@@ -319,6 +344,7 @@ mod tests {
             state.snapshot(),
             WorkerStateSnapshot {
                 last_command_id: 9,
+                last_finished_submission_request_id: None,
                 state: WorkerState::Idle,
             }
         );

crates/worker/tests/controller_test.rs

@@ -1299,12 +1299,10 @@ async fn run_end_returns_to_idle_without_busy_status() {
     let worker = make_worker(client).await;
     let handle = spawn_controller(worker).await;
     let mut rx = handle.subscribe();
+    let submission_request_id = protocol::new_submission_request_id();
 
     handle
-        .send(Method::submit_text(
-            protocol::new_submission_request_id(),
-            "Hello",
-        ))
+        .send(Method::submit_text(submission_request_id.clone(), "Hello"))
         .await
         .unwrap();
 
@@ -1319,7 +1317,11 @@ async fn run_end_returns_to_idle_without_busy_status() {
                         saw_run_end = true;
                     }
                     Ok(Event::WorkerState { snapshot })
-                        if saw_run_end && snapshot.catalog_status() == WorkerStatus::Idle => {
+                        if saw_run_end
+                            && snapshot.catalog_status() == WorkerStatus::Idle
+                            && snapshot.last_finished_submission_request_id.as_deref()
+                                == Some(submission_request_id.as_str()) =>
+                    {
                         saw_idle_status = true;
                         break;
                     }
@@ -1334,7 +1336,7 @@ async fn run_end_returns_to_idle_without_busy_status() {
     assert!(saw_run_end, "expected RunEnd::Finished");
     assert!(
         saw_idle_status,
-        "expected idle status immediately after RunEnd"
+        "expected exact Submit teardown fence in idle status immediately after RunEnd"
     );
     assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
 }