diff --git a/codex-rs/ext/workflows/src/activation.rs b/codex-rs/ext/workflows/src/activation.rs index 9158d03a4..87912e7fd 100644 --- a/codex-rs/ext/workflows/src/activation.rs +++ b/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, } + struct FailingProviderCreditAuthority { + reserve_calls: Arc, + } + #[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 { + 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) diff --git a/codex-rs/ext/workflows/src/manager_tool.rs b/codex-rs/ext/workflows/src/manager_tool.rs index 42b622c17..7835c7b83 100644 --- a/codex-rs/ext/workflows/src/manager_tool.rs +++ b/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) ); } diff --git a/codex-rs/state/src/runtime/workflow_goal_plan_projections.rs b/codex-rs/state/src/runtime/workflow_goal_plan_projections.rs index f5d1b7bbc..9ff7368e9 100644 --- a/codex-rs/state/src/runtime/workflow_goal_plan_projections.rs +++ b/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]