fix(workflows): terminalize pending cancellation projection

codex-rs/ext/workflows/src/activation.rs

@@ -747,6 +747,11 @@ impl WorkflowActivationService {
                 .await?
                 .ok_or_else(|| anyhow::anyhow!("workflow spec record not found"))?;
             let spec = parse_workflow_yaml(spec_record.source_yaml.as_str())?;
+            if claim.snapshot.run.status == WorkflowRunStatus::CancelRequested {
+                self.drive_generation(run_id, claim.generation, &spec, config.clone())
+                    .await?;
+                continue;
+            }
             config.route_runtime.credit_control = self
                 .provider_credit_authority
                 .reserve_or_restore(&fence, &spec, &config.route_runtime)
@@ -2026,6 +2031,10 @@ mod tests {
         reserve_calls: Arc<AtomicUsize>,
     }
 
+    struct FailingProviderCreditAuthority {
+        reserve_calls: Arc<AtomicUsize>,
+    }
+
     #[async_trait]
     impl WorkflowProviderCreditAuthority for FenceAssertingProviderAuthority {
         async fn reserve_or_restore(
@@ -2061,6 +2070,27 @@ mod tests {
         }
     }
 
+    #[async_trait]
+    impl WorkflowProviderCreditAuthority for FailingProviderCreditAuthority {
+        async fn reserve_or_restore(
+            &self,
+            _fence: &WorkflowRunFenceParams,
+            _spec: &WorkflowSpec,
+            _runtime: &WorkflowRouteRuntime,
+        ) -> anyhow::Result<WorkflowProviderCreditControl> {
+            self.reserve_calls.fetch_add(1, Ordering::SeqCst);
+            anyhow::bail!("provider credit reservation unavailable")
+        }
+
+        async fn reconcile_terminal(
+            &self,
+            _fence: &WorkflowRunFenceParams,
+            _status: WorkflowProviderCreditReservationStatus,
+        ) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
     const ROUTE_ACTIVATION_WORKFLOW_YAML: &str = r#"
 schema_version: "workflow.codex.codewith/v0"
 workflow_id: "wf_route_activation"
@@ -2228,6 +2258,144 @@ cleanup:
         assert!(outcome.snapshot.run.owner_id.is_some());
     }
 
+    #[tokio::test]
+    async fn pending_only_cancel_terminalizes_without_provider_reservation_or_parent_plan_mutation()
+    {
+        let temp_dir = tempfile::tempdir().expect("create state home");
+        let state_db = StateRuntime::init(
+            temp_dir.path().to_path_buf(),
+            "cancel-order-test".to_string(),
+        )
+        .await
+        .expect("state db should initialize");
+        let source_thread_id = ThreadId::new();
+        let mut thread = ThreadMetadataBuilder::new(
+            source_thread_id,
+            state_db
+                .codex_home()
+                .join(format!("rollout-{source_thread_id}.jsonl")),
+            Utc::now(),
+            SessionSource::Cli,
+        );
+        thread.cwd = temp_dir.path().to_path_buf();
+        state_db
+            .upsert_thread(&thread.build("cancel-order-test"))
+            .await
+            .expect("source thread should be persisted");
+
+        let parent = state_db
+            .thread_goals()
+            .create_thread_goal_plan(codex_state::ThreadGoalPlanCreateParams {
+                thread_id: source_thread_id,
+                auto_execute: codex_state::ThreadGoalPlanAutoExecute::Off,
+                max_tokens: None,
+                nodes: vec![codex_state::ThreadGoalPlanNodeCreateParams {
+                    key: "parent_payroll_work".to_string(),
+                    objective: "Preserve the unrelated parent payroll goal plan.".to_string(),
+                    assigned_thread_id: None,
+                    title: Some("Parent payroll work".to_string()),
+                    priority: 0,
+                    token_budget: None,
+                    depends_on: Vec::new(),
+                }],
+            })
+            .await
+            .expect("parent goal plan should create")
+            .snapshot;
+        let spec = state_db
+            .workflows()
+            .save_workflow_spec_yaml(WorkflowSpecCreateParams {
+                source_thread_id: Some(source_thread_id),
+                source_yaml: ROUTE_ACTIVATION_WORKFLOW_YAML.replace("  max_tokens: 1000\n", ""),
+            })
+            .await
+            .expect("workflow spec should save");
+        let run = state_db
+            .workflows()
+            .create_workflow_run(WorkflowRunCreateParams {
+                workflow_record_id: spec.workflow_record_id,
+                source_thread_id: Some(source_thread_id),
+                idempotency_key: Some("pending-only-cancel".to_string()),
+            })
+            .await
+            .expect("workflow run should create");
+        let projection = state_db
+            .project_workflow_run_to_goal_plan(WorkflowGoalPlanProjectionParams {
+                workflow_run_id: run.run.run_id.clone(),
+                thread_id: source_thread_id,
+                idempotency_key: Some("pending-only-cancel-projection".to_string()),
+            })
+            .await
+            .expect("workflow projection should succeed")
+            .expect("workflow run should project");
+        assert_eq!(None, projection.snapshot.plan.max_tokens);
+        assert!(
+            projection
+                .snapshot
+                .nodes
+                .iter()
+                .all(|node| node.token_budget.is_none())
+        );
+
+        let reserve_calls = Arc::new(AtomicUsize::new(0));
+        let service = WorkflowActivationService::new_with_provider_credit_authority(
+            Arc::clone(&state_db),
+            Arc::new(FailingProviderCreditAuthority {
+                reserve_calls: Arc::clone(&reserve_calls),
+            }),
+        );
+        let cancelled = tokio::time::timeout(
+            Duration::from_secs(2),
+            service.cancel_workflow_run(
+                WorkflowRunCancelParams {
+                    run_id: run.run.run_id.clone(),
+                    reason: "cancel pending workflow".to_string(),
+                },
+                WorkflowActivationConfig {
+                    route_runtime: supported_route_runtime(),
+                    ..Default::default()
+                },
+            ),
+        )
+        .await
+        .expect("pending-only cancellation should not wait on provider credit")
+        .expect("pending-only cancellation should succeed")
+        .expect("workflow run should exist");
+
+        assert_eq!(0, reserve_calls.load(Ordering::SeqCst));
+        assert_eq!(WorkflowRunStatus::Cancelled, cancelled.run.status);
+        assert!(
+            cancelled
+                .steps
+                .iter()
+                .all(|step| step.status == codex_state::WorkflowRunStepStatus::Cancelled)
+        );
+        let plans = state_db
+            .thread_goals()
+            .list_thread_goal_plans(source_thread_id)
+            .await
+            .expect("goal plans should list");
+        let parent_after = plans
+            .iter()
+            .find(|candidate| candidate.plan.plan_id == parent.plan.plan_id)
+            .expect("parent goal plan should remain");
+        assert_eq!(&parent, parent_after);
+        let projected_after = plans
+            .iter()
+            .find(|candidate| candidate.plan.plan_id == projection.plan_id)
+            .expect("workflow goal plan should remain projected");
+        assert_eq!(
+            codex_state::ThreadGoalPlanStatus::Cancelled,
+            projected_after.plan.status
+        );
+        assert!(
+            projected_after
+                .nodes
+                .iter()
+                .all(|node| node.status == codex_state::ThreadGoalPlanNodeStatus::Cancelled)
+        );
+    }
+
     #[test]
     fn activation_route_gate_accepts_supported_exact_route_and_rejects_before_effects() {
         let spec = parse_workflow_yaml(ROUTE_ACTIVATION_WORKFLOW_YAML)

codex-rs/ext/workflows/src/manager_tool.rs

@@ -1156,6 +1156,25 @@ cleanup:
             .upsert_thread(&thread.build("test-provider"))
             .await
             .expect("thread metadata should insert");
+        let parent = state_db
+            .thread_goals()
+            .create_thread_goal_plan(codex_state::ThreadGoalPlanCreateParams {
+                thread_id,
+                auto_execute: codex_state::ThreadGoalPlanAutoExecute::Off,
+                max_tokens: None,
+                nodes: vec![codex_state::ThreadGoalPlanNodeCreateParams {
+                    key: "unrelated_parent_work".to_string(),
+                    objective: "Preserve this unrelated parent goal plan.".to_string(),
+                    assigned_thread_id: None,
+                    title: Some("Unrelated parent work".to_string()),
+                    priority: 0,
+                    token_budget: None,
+                    depends_on: Vec::new(),
+                }],
+            })
+            .await
+            .expect("unrelated parent goal plan should create")
+            .snapshot;
         let activation_service = Arc::new(
             WorkflowActivationService::new_with_provider_credit_authority(
                 state_db.clone(),
@@ -1197,6 +1216,24 @@ cleanup:
         assert_eq!(start["error"]["recovered"], true);
         assert_eq!(start["run"]["run"]["status"], "cancelled");
         assert_eq!(start["goalPlan"]["status"], "cancelled");
+        assert!(
+            start["run"]["steps"]
+                .as_array()
+                .expect("workflow steps should serialize")
+                .iter()
+                .all(|step| step["status"] == "cancelled")
+        );
+        let replay = call_tool(
+            &tool,
+            json!({
+                "action": "start",
+                "workflow_record_id": create["workflow"]["workflowRecordId"],
+                "idempotency_key": "recoverable-start-failure",
+            }),
+        )
+        .await;
+        assert_eq!(start["run"]["run"]["runId"], replay["run"]["run"]["runId"]);
+        assert_eq!(start["goalPlan"]["planId"], replay["goalPlan"]["planId"]);
         let runs = state_db
             .workflows()
             .list_thread_workflow_runs_page(thread_id, /*cursor*/ None, /*limit*/ 10)
@@ -1208,10 +1245,30 @@ cleanup:
             .list_thread_goal_plans(thread_id)
             .await
             .expect("workflow goal plans should list");
-        assert_eq!(1, plans.len());
+        assert_eq!(2, plans.len());
+        let parent_after = plans
+            .iter()
+            .find(|candidate| candidate.plan.plan_id == parent.plan.plan_id)
+            .expect("unrelated parent goal plan should remain");
+        assert_eq!(&parent, parent_after);
+        let projected_plan_id = start["goalPlan"]["planId"]
+            .as_str()
+            .expect("projected goal plan id should serialize");
+        let projected = plans
+            .iter()
+            .find(|candidate| candidate.plan.plan_id == projected_plan_id)
+            .expect("workflow goal plan should remain projected");
         assert_eq!(
             codex_state::ThreadGoalPlanStatus::Cancelled,
-            plans[0].plan.status
+            projected.plan.status
+        );
+        assert_eq!(None, projected.plan.max_tokens);
+        assert!(
+            projected
+                .nodes
+                .iter()
+                .all(|node| node.token_budget.is_none()
+                    && node.status == codex_state::ThreadGoalPlanNodeStatus::Cancelled)
         );
     }
 

codex-rs/state/src/runtime/workflow_goal_plan_projections.rs

@@ -1043,6 +1043,22 @@ cleanup:
                 .await
                 .expect("goal plan count should read");
         assert_eq!(1, plan_count);
+        let projection_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM workflow_goal_plan_projections WHERE run_id = ?",
+        )
+        .bind(run.run.run_id.as_str())
+        .fetch_one(runtime.thread_goals().pool.as_ref())
+        .await
+        .expect("workflow projection count should read");
+        assert_eq!(1, projection_count);
+        let node_projection_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM workflow_goal_plan_node_projections WHERE run_id = ?",
+        )
+        .bind(run.run.run_id.as_str())
+        .fetch_one(runtime.thread_goals().pool.as_ref())
+        .await
+        .expect("workflow node projection count should read");
+        assert_eq!(2, node_projection_count);
 
         let route_json: String = sqlx::query_scalar(
             r#"
@@ -1087,7 +1103,7 @@ WHERE run_id = ? AND step_id = 'adversarial_review'
 
         let err = runtime
             .project_workflow_run_to_goal_plan(WorkflowGoalPlanProjectionParams {
-                workflow_run_id: run_b.run.run_id,
+                workflow_run_id: run_b.run.run_id.clone(),
                 thread_id: thread_a,
                 idempotency_key: None,
             })
@@ -1102,6 +1118,14 @@ WHERE run_id = ? AND step_id = 'adversarial_review'
                 .await
                 .expect("goal plan count should read");
         assert_eq!(0, plan_count);
+        let projection_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM workflow_goal_plan_projections WHERE run_id = ?",
+        )
+        .bind(run_b.run.run_id.as_str())
+        .fetch_one(runtime.thread_goals().pool.as_ref())
+        .await
+        .expect("workflow projection count should read");
+        assert_eq!(0, projection_count);
     }
 
     #[tokio::test]