Fix managed workflow lifecycle recovery

codex-rs/app-server/src/request_processors/thread_workflow_processor.rs

@@ -378,15 +378,23 @@ impl ThreadWorkflowRequestProcessor {
         if !workflow_run_belongs_to_thread(&state_db, thread_id, params.run_id.as_str()).await? {
             return Ok(ThreadWorkflowRunCancelResponse { run: None });
         }
+        let run_id = params.run_id;
         let cancel_params = codex_state::WorkflowRunCancelParams {
-            run_id: params.run_id,
+            run_id: run_id.clone(),
             reason: params
                 .reason
                 .and_then(normalize_optional_string)
                 .unwrap_or_else(|| "user requested cancellation".to_string()),
         };
-        let run = state_db
-            .request_workflow_run_cancel(cancel_params)
+        let activation_service = self
+            .activation_service
+            .as_ref()
+            .ok_or_else(|| internal_error("workflow activation service is not initialized"))?;
+        let run = activation_service
+            .cancel_workflow_run(
+                cancel_params,
+                crate::extensions::workflow_activation_config(&self.config),
+            )
             .await
             .map_err(|err| internal_error(format!("failed to cancel thread workflow run: {err}")))?
             .map(api_thread_workflow_run_snapshot_from_state);

codex-rs/app-server/tests/suite/v2/workflow.rs

@@ -472,9 +472,38 @@ async fn workflow_run_lifecycle_projects_tasks_and_returns_sanitized_state() ->
     assert_does_not_leak(&serde_json::to_string(&cancel_resp.result)?)?;
     let cancelled = to_response::<ThreadWorkflowRunCancelResponse>(cancel_resp)?;
     assert_eq!(
-        Some(ThreadWorkflowRunStatus::CancelRequested),
+        Some(ThreadWorkflowRunStatus::Cancelled),
         cancelled.run.as_ref().map(|run| run.run.status)
     );
+    let runtime = open_state_runtime(codex_home.path()).await?;
+    let terminal = timeout(DEFAULT_TIMEOUT, async {
+        loop {
+            let snapshot = runtime
+                .workflows()
+                .get_workflow_run_snapshot(started.run.run.run_id.as_str())
+                .await?
+                .ok_or_else(|| anyhow::anyhow!("cancelled workflow run disappeared"))?;
+            if snapshot.run.status.is_terminal() {
+                return Result::<_, anyhow::Error>::Ok(snapshot);
+            }
+            sleep(std::time::Duration::from_millis(50)).await;
+        }
+    })
+    .await??;
+    assert_eq!(
+        codex_state::WorkflowRunStatus::Cancelled,
+        terminal.run.status
+    );
+    let projected_plan = runtime
+        .thread_goals()
+        .list_thread_goal_plans(parse_thread_id(thread_id.as_str())?)
+        .await?
+        .pop()
+        .ok_or_else(|| anyhow::anyhow!("cancelled projected goal plan is missing"))?;
+    assert_eq!(
+        codex_state::ThreadGoalPlanStatus::Cancelled,
+        projected_plan.plan.status
+    );
     assert!(!marker.exists());
 
     Ok(())