fix: persist worker run completion fence

crates/worker-runtime/src/fs_store.rs

@@ -523,6 +523,7 @@ pub(crate) struct PersistedWorkerRecord {
     pub(crate) config_bundle: Option<ConfigBundleRef>,
     pub(crate) created_at_ms: Option<u64>,
     pub(crate) status: WorkerStatus,
+    pub(crate) last_finished_submission_request_id: Option<String>,
     pub(crate) execution_state: PersistedWorkerExecutionState,
     pub(crate) workspace_id: Option<String>,
     pub(crate) workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
@@ -1824,6 +1825,8 @@ struct WorkerAggregateRecord {
     created_at_ms: Option<u64>,
     status: WorkerStatus,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    last_finished_submission_request_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     workspace_id: Option<String>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
@@ -1874,6 +1877,7 @@ impl WorkerAggregateRecord {
             config_bundle: worker.config_bundle.clone(),
             created_at_ms: worker.created_at_ms,
             status: worker.status,
+            last_finished_submission_request_id: worker.last_finished_submission_request_id.clone(),
             workspace_id: worker.workspace_id.clone(),
             workdir_attachments: worker.workdir_attachments.clone(),
             logical_workdir_attachments: worker.logical_workdir_attachments.clone(),
@@ -1893,6 +1897,7 @@ impl WorkerAggregateRecord {
     }
 
     fn validate(self, path: &Path) -> Result<PersistedWorkerRecord, RuntimeError> {
+        let last_finished_submission_request_id = self.last_finished_submission_request_id;
         let identity = WorkerIdentityRecord {
             schema_version: self.schema_version,
             worker_ref: self.worker_ref,
@@ -1984,7 +1989,9 @@ impl WorkerAggregateRecord {
             }
             WorkerExecutionStateRecord::Unavailable => PersistedWorkerExecutionState::Unavailable,
         };
-        Ok(identity.into_persisted(execution_state))
+        let mut persisted = identity.into_persisted(execution_state);
+        persisted.last_finished_submission_request_id = last_finished_submission_request_id;
+        Ok(persisted)
     }
 }
 
@@ -2164,6 +2171,7 @@ impl WorkerIdentityRecord {
             config_bundle: self.config_bundle,
             created_at_ms: self.created_at_ms,
             status: self.status,
+            last_finished_submission_request_id: None,
             execution_state,
             workspace_id: self.workspace_id,
             workdir_attachments,

crates/worker-runtime/src/runtime.rs

@@ -1095,6 +1095,7 @@ impl Runtime {
                 worker_id: worker_id.clone(),
                 status: WorkerStatus::Stopped,
                 worker_state: None,
+                last_finished_submission_request_id: None,
                 workspace_id: scope.map(|scope| scope.workspace_id.clone()),
                 profile: durable_request.profile.clone(),
                 display_name: durable_request.display_name.clone(),
@@ -3419,6 +3420,16 @@ impl Runtime {
         candidate.pending_stop = None;
         candidate.restore_candidate_context = None;
         candidate.status = status;
+        let mut worker_state = worker_state;
+        if worker_state.last_finished_submission_request_id.is_none() {
+            // The restored controller starts with fresh live state, but this
+            // historical fence proves that a previously accepted Submit Run
+            // already returned. Preserve it across process reconstruction so
+            // durable consumers can finish post-restart cleanup without
+            // waiting for an old request to run again.
+            worker_state.last_finished_submission_request_id =
+                candidate.last_finished_submission_request_id.clone();
+        }
         let _ = candidate.apply_worker_state(&worker_state);
         candidate.restore_intent = restore_intent_for_status(candidate.status);
         candidate.workdir_attachments = workdir_attachments;
@@ -4029,6 +4040,7 @@ impl RuntimeState {
                     worker_id: worker.worker_id,
                     status: worker.status,
                     worker_state: None,
+                    last_finished_submission_request_id: worker.last_finished_submission_request_id,
                     workspace_id: worker.workspace_id,
                     profile: worker.profile,
                     display_name: worker.display_name,
@@ -4926,6 +4938,7 @@ struct WorkerRecord {
     worker_id: WorkerId,
     status: WorkerStatus,
     worker_state: Option<protocol::WorkerStateSnapshot>,
+    last_finished_submission_request_id: Option<String>,
     workspace_id: Option<String>,
     profile: ProfileSelector,
     display_name: Option<String>,
@@ -4947,6 +4960,9 @@ struct WorkerRecord {
 
 impl WorkerRecord {
     fn apply_worker_state(&mut self, incoming: &protocol::WorkerStateSnapshot) {
+        if let Some(request_id) = &incoming.last_finished_submission_request_id {
+            self.last_finished_submission_request_id = Some(request_id.clone());
+        }
         self.worker_state = Some(incoming.clone());
     }
 
@@ -5055,6 +5071,7 @@ impl WorkerRecord {
             config_bundle: self.config_bundle.clone(),
             created_at_ms: self.created_at_ms,
             status: self.status,
+            last_finished_submission_request_id: self.last_finished_submission_request_id.clone(),
             execution_state,
             workspace_id: self.workspace_id.clone(),
             workdir_attachments: self.workdir_attachments.clone(),
@@ -9231,6 +9248,75 @@ mod tests {
         let _ = std::fs::remove_dir_all(root);
     }
 
+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn fs_store_preserves_finished_submission_fence_across_active_restore() {
+        let root = fs_store_root("finished-submission-restore");
+        let options = crate::fs_store::FsRuntimeStoreOptions {
+            root: root.clone(),
+            runtime_id: "test-runtime".to_string(),
+            display_name: None,
+        };
+        let execution = Arc::new(TestExecutionBackend::default());
+        let runtime =
+            Runtime::with_fs_store_and_execution_backend(options.clone(), execution.clone())
+                .unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let worker = runtime
+            .create_worker(task_request("persist finished submission fence"))
+            .unwrap();
+        execution.set_dispatch_result(
+            WorkerExecutionResult::accepted_submission(
+                WorkerExecutionOperation::Input,
+                "job-attempt-request",
+                "submission-id",
+                protocol::SubmissionDisposition::Started,
+            )
+            .with_worker_state(protocol::WorkerStateSnapshot {
+                state: protocol::WorkerState::Idle,
+                // Production Submit Run does not consume a lifecycle command
+                // id, so this intentionally remains zero.
+                last_command_id: 0,
+                last_finished_submission_request_id: Some("job-attempt-request".to_string()),
+            }),
+        );
+        let mut input = WorkerInput::user("run durable Backend Job attempt");
+        input.submission_request_id = Some("job-attempt-request".to_string());
+        runtime.send_input(&worker.worker_ref, input).unwrap();
+
+        let aggregate_path = root
+            .join("workers")
+            .join(worker.worker_id.to_string())
+            .join("worker.json");
+        let aggregate: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&aggregate_path).unwrap()).unwrap();
+        assert_eq!(
+            aggregate["last_finished_submission_request_id"],
+            serde_json::json!("job-attempt-request")
+        );
+        drop(runtime);
+
+        let restored = Runtime::with_fs_store_and_execution_backend(
+            options,
+            Arc::new(TestExecutionBackend::default()),
+        )
+        .unwrap();
+        let restored_state = restored
+            .worker_detail(&worker.worker_ref)
+            .unwrap()
+            .worker_state
+            .expect("automatic restore publishes fresh Worker state");
+        assert_eq!(restored_state.last_command_id, 0);
+        assert_eq!(
+            restored_state
+                .last_finished_submission_request_id
+                .as_deref(),
+            Some("job-attempt-request")
+        );
+
+        let _ = std::fs::remove_dir_all(root);
+    }
+
     #[cfg(feature = "fs-store")]
     #[test]
     fn fs_store_explicitly_retries_rejected_automatic_restore() {