From c627f28d37b5cddad4a325ef3ce20e784a06ca30 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 05:12:08 +0300 Subject: [PATCH 01/11] test(workflows): require start to admit execution Agent: vespasian --- codex-rs/ext/workflows/src/manager_tool.rs | 24 +++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/codex-rs/ext/workflows/src/manager_tool.rs b/codex-rs/ext/workflows/src/manager_tool.rs index e2391b1ea9..115a1f2df4 100644 --- a/codex-rs/ext/workflows/src/manager_tool.rs +++ b/codex-rs/ext/workflows/src/manager_tool.rs @@ -637,7 +637,29 @@ mod tests { ) .await; assert_eq!(start["action"], "start"); - assert_eq!(start["run"]["run"]["status"], "pending"); + assert_eq!(start["run"]["run"]["status"], "running"); + assert!( + start["run"]["run"]["activeStepCount"] + .as_i64() + .is_some_and(|count| count > 0), + "starting a workflow must admit at least one ready branch" + ); + let run_id = start["run"]["run"]["runId"] + .as_str() + .expect("started run should include its id"); + let persisted = state_db + .workflows() + .get_workflow_run_snapshot(run_id) + .await + .expect("workflow run should load") + .expect("workflow run should exist"); + assert!( + persisted.steps.iter().any(|step| { + step.status == codex_state::WorkflowRunStepStatus::Active + && step.background_agent_run_id.is_some() + }), + "starting a workflow must queue an independent background worker" + ); assert_eq!( start["goalPlan"]["nodeCount"], start["run"]["run"]["pendingStepCount"] From 3f2052218f1321994c96b4a9e808c26341f70ade Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 05:25:07 +0300 Subject: [PATCH 02/11] fix(workflows): execute admitted workflow branches Agent: vespasian --- codex-rs/app-server/src/message_processor.rs | 1 + codex-rs/app-server/src/request_processors.rs | 1 + .../thread_workflow_processor.rs | 19 + .../thread_workflow_runtime.rs | 535 ++++++++++++++++++ codex-rs/ext/workflows/src/manager_tool.rs | 62 +- codex-rs/state/src/lib.rs | 2 + .../src/runtime/workflow_orchestrator.rs | 192 ++++++- 7 files changed, 808 insertions(+), 4 deletions(-) create mode 100644 codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index cb35840347..bc2e847749 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -599,6 +599,7 @@ impl MessageProcessor { } else { thread_processor.start_background_agent_supervisor(); } + thread_processor.start_workflow_run_supervisor(); } let turn_processor = TurnRequestProcessor::new( auth_manager.clone(), diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 873877c214..71d5e62931 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -689,6 +689,7 @@ mod thread_schedule_default_prompt; mod thread_schedule_processor; mod thread_schedule_runtime; mod thread_workflow_processor; +mod thread_workflow_runtime; mod token_usage_replay; mod turn_processor; mod usage_profile_broker; diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs b/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs index bae026e41e..d7337a9b63 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs @@ -317,6 +317,25 @@ impl ThreadWorkflowRequestProcessor { )) })? .map(|outcome| api_thread_goal_plan_from_state(outcome.snapshot)); + let execution = state_db + .start_workflow_run_execution(codex_state::WorkflowRunStartExecutionParams { + run_id: snapshot.run.run_id.clone(), + owner_id: format!("workflow-manager:{thread_id}"), + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: None, + }) + .await + .map_err(|err| { + internal_error(format!( + "failed to admit workflow execution branches: {err}" + )) + })?; + let snapshot = execution + .map(|execution| execution.snapshot) + .unwrap_or(snapshot); Ok(ThreadWorkflowRunStartResponse { run: api_thread_workflow_run_snapshot_from_state(snapshot), goal_plan, diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs new file mode 100644 index 0000000000..830ec630d0 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs @@ -0,0 +1,535 @@ +use super::thread_processor::ThreadRequestProcessor; +use codex_protocol::protocol::SandboxPolicy; +use codex_state::StateRuntime; +use codex_state::WorkflowRunAdvanceParams; +use codex_state::WorkflowRunBranchAdmissionParams; +use codex_state::WorkflowRunBranchReconcileParams; +use codex_state::WorkflowRunClaimParams; +use codex_state::WorkflowRunStep; +use codex_state::WorkflowRunStepStatus; +use codex_state::WorkflowRunStepVerifier; +use codex_state::WorkflowRunStepVerifierStatus; +use codex_state::WorkflowRunVerifierClaimParams; +use codex_state::WorkflowRunVerifierClaimSelection; +use codex_state::WorkflowRunVerifierOutcomeStatus; +use codex_state::WorkflowRunVerifierRecordResultParams; +use codex_state::WorkflowRunVerifierResultSummary; +use codex_utils_absolute_path::AbsolutePathBuf; +use serde_json::Value; +use std::collections::HashMap; +use std::io; +use std::path::Component; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; +use std::time::Instant; +use tokio::io::AsyncRead; +use tokio::io::AsyncReadExt; +use tracing::warn; + +const WORKFLOW_RUNTIME_RECONCILE_INTERVAL: Duration = Duration::from_secs(5); +const WORKFLOW_RUNTIME_RUN_LIMIT: u32 = 200; + +#[derive(Clone)] +struct WorkflowRuntimeContext { + state_db: Arc, + codex_linux_sandbox_exe: Option, + use_legacy_landlock: bool, +} + +struct VerifierExecution { + outcome: WorkflowRunVerifierOutcomeStatus, + summary: WorkflowRunVerifierResultSummary, +} + +impl ThreadRequestProcessor { + pub(crate) fn start_workflow_run_supervisor(&self) { + let Some(state_db) = self.state_db.clone() else { + return; + }; + let context = WorkflowRuntimeContext { + state_db, + codex_linux_sandbox_exe: self.arg0_paths.codex_linux_sandbox_exe.clone(), + use_legacy_landlock: self.config.features.use_legacy_landlock(), + }; + let cancel_token = self.background_agent_supervisor_token.clone(); + self.background_tasks.spawn(async move { + let mut interval = tokio::time::interval(WORKFLOW_RUNTIME_RECONCILE_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + if let Err(err) = reconcile_workflow_runs(&context).await { + warn!("workflow runtime reconcile failed: {err}"); + } + tokio::select! { + _ = cancel_token.cancelled() => break, + _ = interval.tick() => {} + } + } + }); + } +} + +async fn reconcile_workflow_runs(context: &WorkflowRuntimeContext) -> anyhow::Result<()> { + let run_ids = context + .state_db + .list_active_workflow_run_ids(WORKFLOW_RUNTIME_RUN_LIMIT) + .await?; + for run_id in run_ids { + if let Err(err) = reconcile_workflow_run(context, run_id.as_str()).await { + warn!(run_id, "workflow run reconcile failed: {err}"); + } + } + Ok(()) +} + +async fn reconcile_workflow_run( + context: &WorkflowRuntimeContext, + run_id: &str, +) -> anyhow::Result<()> { + let Some(initial_snapshot) = context + .state_db + .workflows() + .get_workflow_run_snapshot(run_id) + .await? + else { + return Ok(()); + }; + let Some(source_thread_id) = initial_snapshot.run.source_thread_id else { + return Ok(()); + }; + let owner_id = format!("workflow-manager:{source_thread_id}"); + let Some(claim) = context + .state_db + .claim_workflow_run(WorkflowRunClaimParams { + run_id: run_id.to_string(), + owner_id: owner_id.clone(), + lease_duration_ms: None, + }) + .await? + else { + return Ok(()); + }; + let generation = claim.generation; + let Some(reconciled) = context + .state_db + .reconcile_workflow_run_branches(WorkflowRunBranchReconcileParams { + run_id: run_id.to_string(), + owner_id: owner_id.clone(), + generation, + }) + .await? + else { + return Ok(()); + }; + let mut snapshot = reconciled.snapshot; + + loop { + let Some(verifier) = next_executable_verifier(&snapshot) else { + break; + }; + let Some(claimed) = context + .state_db + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run_id.to_string(), + owner_id: owner_id.clone(), + generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId( + verifier.verifier_run_id.clone(), + ), + }) + .await? + else { + break; + }; + let execution = execute_verifier(context, &claimed.step, &claimed.verifier).await; + let Some(recorded) = context + .state_db + .record_workflow_run_verifier_result(WorkflowRunVerifierRecordResultParams { + run_id: run_id.to_string(), + owner_id: owner_id.clone(), + generation, + verifier_run_id: claimed.verifier.verifier_run_id, + outcome: execution.outcome, + summary: execution.summary, + }) + .await? + else { + break; + }; + snapshot = recorded.snapshot; + if execution.outcome == WorkflowRunVerifierOutcomeStatus::Failed { + break; + } + } + + let Some(advanced) = context + .state_db + .advance_workflow_run(WorkflowRunAdvanceParams { + run_id: run_id.to_string(), + owner_id: owner_id.clone(), + generation, + }) + .await? + else { + return Ok(()); + }; + if advanced.snapshot.run.status.is_terminal() { + return Ok(()); + } + context + .state_db + .admit_workflow_run_branches(WorkflowRunBranchAdmissionParams { + run_id: run_id.to_string(), + owner_id, + generation, + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: None, + }) + .await?; + Ok(()) +} + +fn next_executable_verifier( + snapshot: &codex_state::WorkflowRunSnapshot, +) -> Option<&WorkflowRunStepVerifier> { + snapshot.verifiers.iter().find(|verifier| { + matches!( + verifier.status, + WorkflowRunStepVerifierStatus::Pending | WorkflowRunStepVerifierStatus::Blocked + ) && snapshot.steps.iter().any(|step| { + step.step_id == verifier.step_id + && step.status == WorkflowRunStepStatus::WaitingVerifier + }) + }) +} + +async fn execute_verifier( + context: &WorkflowRuntimeContext, + step: &WorkflowRunStep, + verifier: &WorkflowRunStepVerifier, +) -> VerifierExecution { + let started_at = Instant::now(); + let result = execute_verifier_inner(context, step, verifier).await; + match result { + Ok(execution) => execution, + Err(_) => VerifierExecution { + outcome: WorkflowRunVerifierOutcomeStatus::Failed, + summary: WorkflowRunVerifierResultSummary { + command_count: 0, + expected_exit_code: verifier_definition(verifier) + .get("expected_exit_code") + .and_then(Value::as_i64) + .and_then(|value| i32::try_from(value).ok()), + observed_exit_code: None, + timed_out: false, + duration_ms: duration_millis(started_at.elapsed()), + output_bytes: 0, + output_truncated: false, + }, + }, + } +} + +async fn execute_verifier_inner( + context: &WorkflowRuntimeContext, + step: &WorkflowRunStep, + verifier: &WorkflowRunStepVerifier, +) -> anyhow::Result { + let background_agent_run_id = step + .background_agent_run_id + .as_deref() + .ok_or_else(|| anyhow::anyhow!("workflow step has no background agent run"))?; + let status_snapshot = context + .state_db + .get_background_agent_status_snapshot(background_agent_run_id) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow branch status snapshot is missing"))?; + let workspace_cwd = status_snapshot + .payload_json + .get("cwd") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("workflow branch status snapshot has no cwd"))?; + let workspace_cwd = canonical_directory(Path::new(workspace_cwd))?; + match verifier.verifier_type.as_str() { + "artifact_contains" => execute_artifact_verifier(&workspace_cwd, verifier).await, + "run_commands" => execute_command_verifier(context, &workspace_cwd, verifier).await, + verifier_type => anyhow::bail!("unsupported workflow verifier type `{verifier_type}`"), + } +} + +async fn execute_artifact_verifier( + workspace_cwd: &Path, + verifier: &WorkflowRunStepVerifier, +) -> anyhow::Result { + let started_at = Instant::now(); + let definition = verifier_definition(verifier); + let artifact = definition + .get("artifact") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("artifact verifier is missing artifact"))?; + let artifact_path = resolve_beneath(workspace_cwd, artifact)?; + let bytes = tokio::fs::read(&artifact_path).await?; + let content = String::from_utf8_lossy(&bytes); + let passed = definition + .get("must_contain") + .and_then(Value::as_array) + .is_some_and(|needles| { + !needles.is_empty() + && needles + .iter() + .filter_map(Value::as_str) + .all(|needle| content.contains(needle)) + }); + Ok(VerifierExecution { + outcome: if passed { + WorkflowRunVerifierOutcomeStatus::Passed + } else { + WorkflowRunVerifierOutcomeStatus::Failed + }, + summary: WorkflowRunVerifierResultSummary { + command_count: 0, + expected_exit_code: None, + observed_exit_code: None, + timed_out: false, + duration_ms: duration_millis(started_at.elapsed()), + output_bytes: i64::try_from(bytes.len()).unwrap_or(i64::MAX), + output_truncated: false, + }, + }) +} + +async fn execute_command_verifier( + context: &WorkflowRuntimeContext, + workspace_cwd: &Path, + verifier: &WorkflowRunStepVerifier, +) -> anyhow::Result { + let started_at = Instant::now(); + let definition = verifier_definition(verifier); + let verifier_cwd = definition + .get("cwd") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("command verifier is missing cwd"))?; + let verifier_cwd = resolve_beneath(workspace_cwd, verifier_cwd)?; + let timeout_seconds = definition + .get("timeout_seconds") + .and_then(Value::as_u64) + .ok_or_else(|| anyhow::anyhow!("command verifier is missing timeout_seconds"))?; + let output_limit_bytes = definition + .get("output_limit_bytes") + .and_then(Value::as_u64) + .ok_or_else(|| anyhow::anyhow!("command verifier is missing output_limit_bytes"))?; + let expected_exit_code = definition + .get("expected_exit_code") + .and_then(Value::as_i64) + .and_then(|value| i32::try_from(value).ok()) + .unwrap_or(0); + let commands = definition + .get("commands") + .and_then(Value::as_array) + .ok_or_else(|| anyhow::anyhow!("command verifier is missing commands"))?; + let permission_profile = verifier_permission_profile(definition, &verifier_cwd)?; + let absolute_cwd = AbsolutePathBuf::try_from(verifier_cwd.clone())?; + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_seconds); + let mut command_count = 0_i64; + let mut observed_exit_code = None; + let mut output_bytes = 0_i64; + let mut output_truncated = false; + let mut timed_out = false; + + for command in commands.iter().filter_map(Value::as_str) { + command_count += 1; + let mut child = codex_core::exec::spawn_streaming_command_under_sandbox( + vec![ + "/bin/bash".to_string(), + "-lc".to_string(), + command.to_string(), + ], + absolute_cwd.clone(), + verifier_environment(), + &permission_profile, + &absolute_cwd, + &context.codex_linux_sandbox_exe, + context.use_legacy_landlock, + ) + .await?; + let stdout = child + .stdout + .take() + .ok_or_else(|| anyhow::anyhow!("verifier stdout was not piped"))?; + let stderr = child + .stderr + .take() + .ok_or_else(|| anyhow::anyhow!("verifier stderr was not piped"))?; + let stdout_task = tokio::spawn(drain_stream(stdout, output_limit_bytes)); + let stderr_task = tokio::spawn(drain_stream(stderr, output_limit_bytes)); + match tokio::time::timeout_at(deadline, child.wait()).await { + Ok(status) => { + observed_exit_code = status?.code(); + } + Err(_) => { + timed_out = true; + child.kill().await?; + let _ = child.wait().await; + } + } + let stdout_result = stdout_task.await??; + let stderr_result = stderr_task.await??; + output_bytes = output_bytes + .saturating_add(stdout_result.0) + .saturating_add(stderr_result.0); + output_truncated |= stdout_result.1 || stderr_result.1; + if timed_out || observed_exit_code != Some(expected_exit_code) { + break; + } + } + + Ok(VerifierExecution { + outcome: if !timed_out + && command_count == i64::try_from(commands.len())? + && observed_exit_code == Some(expected_exit_code) + { + WorkflowRunVerifierOutcomeStatus::Passed + } else { + WorkflowRunVerifierOutcomeStatus::Failed + }, + summary: WorkflowRunVerifierResultSummary { + command_count, + expected_exit_code: Some(expected_exit_code), + observed_exit_code, + timed_out, + duration_ms: duration_millis(started_at.elapsed()), + output_bytes, + output_truncated, + }, + }) +} + +fn verifier_definition(verifier: &WorkflowRunStepVerifier) -> &Value { + verifier + .definition_json + .get("data") + .unwrap_or(&verifier.definition_json) +} + +fn verifier_permission_profile( + definition: &Value, + cwd: &Path, +) -> anyhow::Result { + let network_access = match definition + .get("network") + .and_then(Value::as_str) + .unwrap_or("disabled") + { + "disabled" | "default" => false, + "enabled" => true, + network => anyhow::bail!("unsupported verifier network policy `{network}`"), + }; + let sandbox_policy = match definition + .get("sandbox") + .and_then(Value::as_str) + .unwrap_or("read-only") + { + "default" | "read-only" => SandboxPolicy::ReadOnly { network_access }, + "workspace-write" => SandboxPolicy::WorkspaceWrite { + writable_roots: Vec::new(), + network_access, + exclude_tmpdir_env_var: false, + exclude_slash_tmp: false, + }, + sandbox => anyhow::bail!("unsupported verifier sandbox policy `{sandbox}`"), + }; + let cwd = AbsolutePathBuf::try_from(cwd.to_path_buf())?; + Ok( + codex_protocol::models::PermissionProfile::from_legacy_sandbox_policy_for_cwd( + &sandbox_policy, + &cwd, + ), + ) +} + +fn verifier_environment() -> HashMap { + ["HOME", "PATH", "LANG", "LC_ALL", "TERM"] + .into_iter() + .filter_map(|key| { + std::env::var(key) + .ok() + .map(|value| (key.to_string(), value)) + }) + .collect() +} + +fn canonical_directory(path: &Path) -> anyhow::Result { + let path = std::fs::canonicalize(path)?; + if !path.is_dir() { + anyhow::bail!("workflow verifier cwd is not a directory"); + } + Ok(path) +} + +fn resolve_beneath(root: &Path, relative: &str) -> anyhow::Result { + let relative_path = Path::new(relative); + if relative_path.is_absolute() + || relative_path.components().any(|component| { + matches!( + component, + Component::ParentDir | Component::RootDir | Component::Prefix(_) + ) + }) + { + anyhow::bail!("workflow verifier path must stay beneath the branch workspace"); + } + let candidate = std::fs::canonicalize(root.join(relative_path))?; + if !candidate.starts_with(root) { + anyhow::bail!("workflow verifier path escaped the branch workspace"); + } + Ok(candidate) +} + +async fn drain_stream( + mut stream: impl AsyncRead + Unpin, + output_limit_bytes: u64, +) -> io::Result<(i64, bool)> { + let mut buffer = [0_u8; 8192]; + let mut total = 0_u64; + loop { + let read = stream.read(&mut buffer).await?; + if read == 0 { + break; + } + total = total.saturating_add(u64::try_from(read).unwrap_or(u64::MAX)); + } + Ok(( + i64::try_from(total).unwrap_or(i64::MAX), + total > output_limit_bytes, + )) +} + +fn duration_millis(duration: Duration) -> i64 { + i64::try_from(duration.as_millis()).unwrap_or(i64::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn verifier_paths_cannot_escape_workspace() { + let root = tempfile::tempdir().expect("tempdir"); + assert!(resolve_beneath(root.path(), "../escape").is_err()); + assert!(resolve_beneath(root.path(), "/tmp").is_err()); + } + + #[test] + fn verifier_permission_profile_rejects_unknown_policy() { + let cwd = tempfile::tempdir().expect("tempdir"); + let definition = serde_json::json!({ + "sandbox": "unsafe", + "network": "disabled", + }); + assert!(verifier_permission_profile(&definition, cwd.path()).is_err()); + } +} diff --git a/codex-rs/ext/workflows/src/manager_tool.rs b/codex-rs/ext/workflows/src/manager_tool.rs index 115a1f2df4..5d5e931b6d 100644 --- a/codex-rs/ext/workflows/src/manager_tool.rs +++ b/codex-rs/ext/workflows/src/manager_tool.rs @@ -327,16 +327,31 @@ impl ManageWorkflowTool { .await .map_err(|_| respond("failed to start workflow run"))?; let run_id = snapshot.run.run_id.clone(); - let run = run_snapshot_json(&snapshot); let goal_plan = state_db .project_workflow_run_to_goal_plan(codex_state::WorkflowGoalPlanProjectionParams { - workflow_run_id: run_id, + workflow_run_id: run_id.clone(), thread_id, idempotency_key, }) .await .map_err(|_| respond("failed to project workflow run into task plan"))? .map(goal_plan_projection_json); + let execution = state_db + .start_workflow_run_execution(codex_state::WorkflowRunStartExecutionParams { + run_id, + owner_id: format!("workflow-manager:{thread_id}"), + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: None, + }) + .await + .map_err(|_| respond("failed to admit workflow execution branches"))?; + let run = execution + .as_ref() + .map(|execution| run_snapshot_json(&execution.snapshot)) + .unwrap_or_else(|| run_snapshot_json(&snapshot)); Ok(json!({ "action": "start", "run": run, @@ -631,7 +646,7 @@ mod tests { &tool, json!({ "action": "start", - "workflow_record_id": workflow_record_id, + "workflow_record_id": workflow_record_id.clone(), "idempotency_key": " run-1 ", }), ) @@ -660,6 +675,47 @@ mod tests { }), "starting a workflow must queue an independent background worker" ); + assert_eq!( + state_db + .get_thread_goal(thread_id) + .await + .expect("source goal should load"), + None, + "workflow execution must not replace the source thread goal" + ); + let first_branch_ids = persisted + .steps + .iter() + .filter_map(|step| step.background_agent_run_id.clone()) + .collect::>(); + let duplicate_start = call_tool( + &tool, + json!({ + "action": "start", + "workflow_record_id": workflow_record_id, + "idempotency_key": "run-1", + }), + ) + .await; + let duplicate_run_id = duplicate_start["run"]["run"]["runId"] + .as_str() + .expect("duplicate start should include its run id"); + assert_eq!(duplicate_run_id, run_id); + let duplicate_persisted = state_db + .workflows() + .get_workflow_run_snapshot(duplicate_run_id) + .await + .expect("duplicate workflow run should load") + .expect("duplicate workflow run should exist"); + assert_eq!( + duplicate_persisted + .steps + .iter() + .filter_map(|step| step.background_agent_run_id.clone()) + .collect::>(), + first_branch_ids, + "duplicate start must not queue duplicate workflow branches" + ); assert_eq!( start["goalPlan"]["nodeCount"], start["run"]["run"]["pendingStepCount"] diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 049d600741..037b050726 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -276,6 +276,8 @@ pub use runtime::WorkflowRunCreateParams; pub use runtime::WorkflowRunListPage; pub use runtime::WorkflowRunPauseParams; pub use runtime::WorkflowRunResumeParams; +pub use runtime::WorkflowRunStartExecutionOutcome; +pub use runtime::WorkflowRunStartExecutionParams; pub use runtime::WorkflowRunStatusMutationOutcome; pub use runtime::WorkflowRunStepApprovalDecision; pub use runtime::WorkflowRunStepApprovalOutcome; diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index 1b89360601..5eba8b21bd 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -104,7 +104,181 @@ pub struct WorkflowRunBranchReconcileOutcome { pub changed: bool, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowRunStartExecutionParams { + pub run_id: String, + pub owner_id: String, + pub auth_profile_ref: Option, + pub config_fingerprint: Option, + pub version_fingerprint: Option, + pub parent_agent_run_id: Option, + pub max_active_background_agent_runs: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowRunStartExecutionOutcome { + pub snapshot: crate::WorkflowRunSnapshot, + pub generation: i64, + pub lease_expires_at_ms: i64, + pub admitted: Vec, +} + impl StateRuntime { + pub async fn list_active_workflow_run_ids(&self, limit: u32) -> anyhow::Result> { + let limit = i64::from(limit.clamp(1, 500)); + sqlx::query_scalar( + r#" +SELECT run_id +FROM workflow_runs +WHERE status NOT IN ('completed', 'failed', 'cancelled', 'paused') +ORDER BY updated_at_ms, run_id +LIMIT ? + "#, + ) + .bind(limit) + .fetch_all(self.pool.as_ref()) + .await + .map_err(anyhow::Error::from) + } + + /// Atomically claims a newly created workflow run, marks dependency-free steps ready, + /// and admits their independent background-agent branches. + pub async fn start_workflow_run_execution( + &self, + params: WorkflowRunStartExecutionParams, + ) -> anyhow::Result> { + self.start_workflow_run_execution_with_provider_env_check(params, provider_env_key_present) + .await + } + + async fn start_workflow_run_execution_with_provider_env_check( + &self, + params: WorkflowRunStartExecutionParams, + provider_env_key_present: impl Fn(&str) -> bool, + ) -> anyhow::Result> { + validate_owner_id(¶ms.owner_id)?; + if params + .max_active_background_agent_runs + .is_some_and(|limit| limit <= 0) + { + anyhow::bail!("max_active_background_agent_runs must be positive when set"); + } + let lease_duration_ms = DEFAULT_WORKFLOW_LEASE_DURATION_MS; + let now_ms = datetime_to_epoch_millis(Utc::now()); + let lease_expires_at_ms = now_ms.saturating_add(lease_duration_ms); + let mut tx = self.pool.begin().await?; + let row = sqlx::query( + r#" +UPDATE workflow_runs +SET + owner_id = ?, + lease_expires_at_ms = ?, + heartbeat_at_ms = ?, + generation = generation + 1, + status = CASE + WHEN status IN ('pending', 'waiting') THEN ? + ELSE status + END, + started_at_ms = COALESCE(started_at_ms, ?), + updated_at_ms = ? +WHERE run_id = ? + AND status NOT IN ('completed', 'failed', 'cancelled', 'paused') + AND ( + owner_id IS NULL + OR owner_id = ? + OR lease_expires_at_ms IS NULL + OR lease_expires_at_ms <= ? + ) +RETURNING generation + "#, + ) + .bind(params.owner_id.as_str()) + .bind(lease_expires_at_ms) + .bind(now_ms) + .bind(crate::WorkflowRunStatus::Running.as_str()) + .bind(now_ms) + .bind(now_ms) + .bind(params.run_id.as_str()) + .bind(params.owner_id.as_str()) + .bind(now_ms) + .fetch_optional(&mut *tx) + .await?; + let Some(row) = row else { + tx.commit().await?; + return Ok(None); + }; + let generation: i64 = row.try_get("generation")?; + append_workflow_run_event_in_tx( + &mut tx, + params.run_id.as_str(), + WorkflowRunEventAppend { + event_type: "claimed", + actor_kind: "orchestrator", + actor_id: Some(params.owner_id.clone()), + step_run_id: None, + verifier_run_id: None, + visibility: "internal", + payload: json!({ + "generation": generation, + "leaseExpiresAtMs": lease_expires_at_ms, + }), + now_ms, + }, + ) + .await?; + + mark_ready_steps_in_tx( + &mut tx, + params.run_id.as_str(), + params.owner_id.as_str(), + now_ms, + ) + .await?; + let run = snapshot_workflow_run_in_tx(&mut tx, params.run_id.as_str()) + .await? + .run; + let admission_params = WorkflowRunBranchAdmissionParams { + run_id: params.run_id.clone(), + owner_id: params.owner_id.clone(), + generation, + auth_profile_ref: params.auth_profile_ref, + config_fingerprint: params.config_fingerprint, + version_fingerprint: params.version_fingerprint, + parent_agent_run_id: params.parent_agent_run_id, + max_active_background_agent_runs: params.max_active_background_agent_runs, + }; + let admission = admit_ready_workflow_branches_in_tx( + &mut tx, + &run, + &admission_params, + &provider_env_key_present, + now_ms, + ) + .await?; + if admission.changed { + recompute_workflow_run_status_in_tx( + &mut tx, + params.run_id.as_str(), + params.owner_id.as_str(), + now_ms, + ) + .await?; + } + let snapshot = snapshot_workflow_run_in_tx(&mut tx, params.run_id.as_str()).await?; + tx.commit().await?; + if admission.blocked_by_provider_preflight { + self.thread_goals + .block_workflow_goal_plan_projection(params.run_id.as_str()) + .await?; + } + Ok(Some(WorkflowRunStartExecutionOutcome { + snapshot, + generation, + lease_expires_at_ms, + admitted: admission.admitted, + })) + } + pub async fn claim_workflow_run( &self, params: WorkflowRunClaimParams, @@ -656,10 +830,13 @@ SELECT step.step_run_id, step.step_id, step.background_agent_run_id, - agent.status + agent.status, + snapshot.payload_json AS status_payload_json FROM workflow_run_steps step JOIN background_agent_runs agent ON agent.id = step.background_agent_run_id +LEFT JOIN background_agent_status_snapshots snapshot + ON snapshot.run_id = agent.id WHERE step.run_id = ? AND step.status = 'active' AND step.background_agent_run_id IS NOT NULL @@ -677,6 +854,10 @@ ORDER BY step.sequence, step.step_id step_id: row.try_get("step_id")?, background_agent_run_id: row.try_get("background_agent_run_id")?, status: row.try_get("status")?, + status_payload_json: row + .try_get::, _>("status_payload_json")? + .map(|payload| serde_json::from_str(payload.as_str())) + .transpose()?, }; if branch.status == BackgroundAgentRunStatus::Completed.as_str() { changed |= mark_branch_completed_in_tx(tx, run_id, owner_id, &branch, now_ms).await?; @@ -730,6 +911,14 @@ WHERE step_run_id = ? payload: json!({ "stepId": branch.step_id.as_str(), "backgroundAgentRunId": branch.background_agent_run_id.as_str(), + "terminalResult": branch + .status_payload_json + .as_ref() + .and_then(|payload| payload.get("finalResult")), + "workspaceCwd": branch + .status_payload_json + .as_ref() + .and_then(|payload| payload.get("cwd")), }), now_ms, }, @@ -845,6 +1034,7 @@ struct TerminalWorkflowBranch { step_id: String, background_agent_run_id: String, status: String, + status_payload_json: Option, } struct WorkflowRunBranchAdmissionTxOutcome { From 03fc12c31d1f6d75342905f4474dfe8e16472f2e Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 05:29:11 +0300 Subject: [PATCH 03/11] test(workflows): preserve source goal on start Agent: vespasian --- codex-rs/ext/workflows/src/manager_tool.rs | 12 +++++++++++- codex-rs/state/src/runtime.rs | 2 ++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/codex-rs/ext/workflows/src/manager_tool.rs b/codex-rs/ext/workflows/src/manager_tool.rs index 5d5e931b6d..1878c8d722 100644 --- a/codex-rs/ext/workflows/src/manager_tool.rs +++ b/codex-rs/ext/workflows/src/manager_tool.rs @@ -623,6 +623,16 @@ mod tests { ) .await .expect("thread metadata should insert"); + let source_goal = state_db + .insert_thread_goal( + thread_id, + "keep the source goal active", + codex_state::ThreadGoalStatus::Active, + None, + ) + .await + .expect("source goal should insert") + .expect("source goal should be active"); let tool = ManageWorkflowTool::new(Arc::new(AtomicBool::new(true)), state_db.clone(), thread_id); @@ -680,7 +690,7 @@ mod tests { .get_thread_goal(thread_id) .await .expect("source goal should load"), - None, + Some(source_goal), "workflow execution must not replace the source thread goal" ); let first_branch_ids = persisted diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 49739f1c84..f1dedbd7d2 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -280,6 +280,8 @@ pub use workflow_orchestrator::WorkflowRunBranchReconcileOutcome; pub use workflow_orchestrator::WorkflowRunBranchReconcileParams; pub use workflow_orchestrator::WorkflowRunClaimOutcome; pub use workflow_orchestrator::WorkflowRunClaimParams; +pub use workflow_orchestrator::WorkflowRunStartExecutionOutcome; +pub use workflow_orchestrator::WorkflowRunStartExecutionParams; pub use workflow_verifiers::WorkflowRunVerifierClaimOutcome; pub use workflow_verifiers::WorkflowRunVerifierClaimParams; pub use workflow_verifiers::WorkflowRunVerifierClaimSelection; From 0102ab626e4f6a244b13d7d4d2d730a931ae0273 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 05:37:08 +0300 Subject: [PATCH 04/11] fix(workflows): activate deterministic verifiers Agent: vespasian --- .../thread_workflow_runtime.rs | 2 +- codex-rs/ext/workflows/src/manager_tool.rs | 6 +- .../src/runtime/workflow_orchestrator.rs | 83 +++---------------- 3 files changed, 16 insertions(+), 75 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs index 830ec630d0..08871b5008 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs @@ -372,7 +372,7 @@ async fn execute_command_verifier( } Err(_) => { timed_out = true; - child.kill().await?; + codex_utils_pty::process_group::kill_child_process_group(&mut child)?; let _ = child.wait().await; } } diff --git a/codex-rs/ext/workflows/src/manager_tool.rs b/codex-rs/ext/workflows/src/manager_tool.rs index 1878c8d722..99b1602722 100644 --- a/codex-rs/ext/workflows/src/manager_tool.rs +++ b/codex-rs/ext/workflows/src/manager_tool.rs @@ -624,6 +624,7 @@ mod tests { .await .expect("thread metadata should insert"); let source_goal = state_db + .thread_goals() .insert_thread_goal( thread_id, "keep the source goal active", @@ -687,6 +688,7 @@ mod tests { ); assert_eq!( state_db + .thread_goals() .get_thread_goal(thread_id) .await .expect("source goal should load"), @@ -727,8 +729,8 @@ mod tests { "duplicate start must not queue duplicate workflow branches" ); assert_eq!( - start["goalPlan"]["nodeCount"], - start["run"]["run"]["pendingStepCount"] + start["goalPlan"]["nodeCount"].as_u64(), + u64::try_from(persisted.steps.len()).ok() ); let run_id = start["run"]["run"]["runId"] .as_str() diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index 5eba8b21bd..385a86e832 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -23,8 +23,8 @@ use uuid::Uuid; const DEFAULT_WORKFLOW_LEASE_DURATION_MS: i64 = 60_000; const OPENROUTER_API_KEY_ENV_VAR: &str = "OPENROUTER_API_KEY"; const OPENROUTER_PROVIDER_ID: &str = "openrouter"; -const VERIFIER_EXECUTOR_PENDING_REASON: &str = "deterministic verifier executor is not enabled"; -const VERIFIER_EXECUTOR_PENDING_REASON_CODE: &str = "verifier_executor_pending"; +const VERIFIER_READY_REASON: &str = "deterministic verifier is ready"; +const VERIFIER_READY_REASON_CODE: &str = "verifier_ready"; const WORKFLOW_BRANCH_ADMITTED_REASON: &str = "workflow branch admitted"; const WORKFLOW_BRANCH_ADMITTED_REASON_CODE: &str = "workflow_branch_admitted"; const WORKFLOW_BRANCH_PROVIDER_ENV_MISSING_REASON: &str = @@ -888,8 +888,8 @@ WHERE step_run_id = ? "#, ) .bind(crate::WorkflowRunStepStatus::WaitingVerifier.as_str()) - .bind(VERIFIER_EXECUTOR_PENDING_REASON) - .bind(VERIFIER_EXECUTOR_PENDING_REASON_CODE) + .bind(VERIFIER_READY_REASON) + .bind(VERIFIER_READY_REASON_CODE) .bind(now_ms) .bind(branch.step_run_id.as_str()) .execute(&mut **tx) @@ -924,8 +924,6 @@ WHERE step_run_id = ? }, ) .await?; - block_pending_step_verifiers_in_tx(tx, run_id, branch.step_id.as_str(), owner_id, now_ms) - .await?; Ok(true) } @@ -1612,8 +1610,8 @@ RETURNING step_run_id "#, ) .bind(crate::WorkflowRunStepStatus::WaitingVerifier.as_str()) - .bind(VERIFIER_EXECUTOR_PENDING_REASON) - .bind(VERIFIER_EXECUTOR_PENDING_REASON_CODE) + .bind(VERIFIER_READY_REASON) + .bind(VERIFIER_READY_REASON_CODE) .bind(now_ms) .bind(run_id) .bind(step_id.as_str()) @@ -1636,75 +1634,16 @@ RETURNING step_run_id visibility: "internal", payload: json!({ "stepId": step_id, - "reasonCode": VERIFIER_EXECUTOR_PENDING_REASON_CODE, + "reasonCode": VERIFIER_READY_REASON_CODE, }), now_ms, }, ) .await?; - changed |= - block_pending_step_verifiers_in_tx(tx, run_id, step_id, owner_id, now_ms).await?; } Ok(changed) } -async fn block_pending_step_verifiers_in_tx( - tx: &mut sqlx::Transaction<'_, Sqlite>, - run_id: &str, - step_id: &str, - owner_id: &str, - now_ms: i64, -) -> anyhow::Result { - let rows = sqlx::query( - r#" -UPDATE workflow_run_step_verifiers -SET - status = ?, - status_reason = ?, - reason_code = ?, - updated_at_ms = ? -WHERE run_id = ? - AND step_id = ? - AND status = 'pending' -RETURNING verifier_run_id, verifier_id, verifier_type - "#, - ) - .bind(crate::WorkflowRunStepVerifierStatus::Blocked.as_str()) - .bind(VERIFIER_EXECUTOR_PENDING_REASON) - .bind(VERIFIER_EXECUTOR_PENDING_REASON_CODE) - .bind(now_ms) - .bind(run_id) - .bind(step_id) - .fetch_all(&mut **tx) - .await?; - for row in &rows { - let verifier_run_id: String = row.try_get("verifier_run_id")?; - let verifier_id: String = row.try_get("verifier_id")?; - let verifier_type: String = row.try_get("verifier_type")?; - append_workflow_run_event_in_tx( - tx, - run_id, - WorkflowRunEventAppend { - event_type: "verifier_blocked", - actor_kind: "orchestrator", - actor_id: Some(owner_id.to_string()), - step_run_id: None, - verifier_run_id: Some(verifier_run_id), - visibility: "internal", - payload: json!({ - "stepId": step_id, - "verifierId": verifier_id, - "verifierType": verifier_type, - "reasonCode": VERIFIER_EXECUTOR_PENDING_REASON_CODE, - }), - now_ms, - }, - ) - .await?; - } - Ok(!rows.is_empty()) -} - async fn promote_verified_steps_in_tx( tx: &mut sqlx::Transaction<'_, Sqlite>, run_id: &str, @@ -2335,8 +2274,8 @@ fn workflow_run_reason_for_status( ) -> (Option<&'static str>, Option<&'static str>) { match status { crate::WorkflowRunStatus::Waiting => ( - Some(VERIFIER_EXECUTOR_PENDING_REASON), - Some(VERIFIER_EXECUTOR_PENDING_REASON_CODE), + Some(VERIFIER_READY_REASON), + Some(VERIFIER_READY_REASON_CODE), ), crate::WorkflowRunStatus::Blocked => { (Some("workflow run is blocked"), Some("workflow_blocked")) @@ -2834,7 +2773,7 @@ WHERE plan_id = ? AND key = ? .find(|verifier| verifier.step_id == "adversarial_scope") .expect("scope verifier should exist"); assert_eq!( - crate::WorkflowRunStepVerifierStatus::Blocked, + crate::WorkflowRunStepVerifierStatus::Pending, scope_verifier.status ); assert!( @@ -3718,7 +3657,7 @@ WHERE run_id = ? reconciled.snapshot.steps[0].status ); assert_eq!( - crate::WorkflowRunStepVerifierStatus::Blocked, + crate::WorkflowRunStepVerifierStatus::Pending, reconciled.snapshot.verifiers[0].status ); } From 2e3f4ff1726e0de3f0ffe2bcd100464094de6562 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 05:54:21 +0300 Subject: [PATCH 05/11] test(workflows): cover executor lifecycle Agent: vespasian --- .../thread_workflow_runtime.rs | 222 ++++++++- .../src/runtime/workflow_orchestrator.rs | 436 +++++++++++++++++- 2 files changed, 625 insertions(+), 33 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs index 08871b5008..8debd0cabb 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs @@ -30,6 +30,7 @@ use tracing::warn; const WORKFLOW_RUNTIME_RECONCILE_INTERVAL: Duration = Duration::from_secs(5); const WORKFLOW_RUNTIME_RUN_LIMIT: u32 = 200; +const MAX_VERIFIER_CAPTURE_BYTES: u64 = 1024 * 1024; #[derive(Clone)] struct WorkflowRuntimeContext { @@ -43,6 +44,11 @@ struct VerifierExecution { summary: WorkflowRunVerifierResultSummary, } +struct DrainedStream { + captured: Vec, + total_bytes: u64, +} + impl ThreadRequestProcessor { pub(crate) fn start_workflow_run_supervisor(&self) { let Some(state_db) = self.state_db.clone() else { @@ -322,6 +328,7 @@ async fn execute_command_verifier( .get("output_limit_bytes") .and_then(Value::as_u64) .ok_or_else(|| anyhow::anyhow!("command verifier is missing output_limit_bytes"))?; + let expected_stdout = definition.get("expected_stdout").and_then(Value::as_str); let expected_exit_code = definition .get("expected_exit_code") .and_then(Value::as_i64) @@ -339,6 +346,11 @@ async fn execute_command_verifier( let mut output_bytes = 0_i64; let mut output_truncated = false; let mut timed_out = false; + let mut stdout_bytes = 0_u64; + let mut captured_stdout = Vec::new(); + let stdout_capture_limit = expected_stdout + .map(|_| output_limit_bytes.min(MAX_VERIFIER_CAPTURE_BYTES)) + .unwrap_or_default(); for command in commands.iter().filter_map(Value::as_str) { command_count += 1; @@ -349,7 +361,7 @@ async fn execute_command_verifier( command.to_string(), ], absolute_cwd.clone(), - verifier_environment(), + verifier_environment(workspace_cwd), &permission_profile, &absolute_cwd, &context.codex_linux_sandbox_exe, @@ -364,8 +376,8 @@ async fn execute_command_verifier( .stderr .take() .ok_or_else(|| anyhow::anyhow!("verifier stderr was not piped"))?; - let stdout_task = tokio::spawn(drain_stream(stdout, output_limit_bytes)); - let stderr_task = tokio::spawn(drain_stream(stderr, output_limit_bytes)); + let stdout_task = tokio::spawn(drain_stream(stdout, stdout_capture_limit)); + let stderr_task = tokio::spawn(drain_stream(stderr, /* capture_limit_bytes */ 0)); match tokio::time::timeout_at(deadline, child.wait()).await { Ok(status) => { observed_exit_code = status?.code(); @@ -378,20 +390,36 @@ async fn execute_command_verifier( } let stdout_result = stdout_task.await??; let stderr_result = stderr_task.await??; - output_bytes = output_bytes - .saturating_add(stdout_result.0) - .saturating_add(stderr_result.0); - output_truncated |= stdout_result.1 || stderr_result.1; + stdout_bytes = stdout_bytes.saturating_add(stdout_result.total_bytes); + let remaining_capture = stdout_capture_limit + .saturating_sub(u64::try_from(captured_stdout.len()).unwrap_or(u64::MAX)); + let append_len = usize::try_from(remaining_capture) + .unwrap_or(usize::MAX) + .min(stdout_result.captured.len()); + captured_stdout.extend_from_slice(&stdout_result.captured[..append_len]); + let command_output_bytes = stdout_result + .total_bytes + .saturating_add(stderr_result.total_bytes); + output_bytes = + output_bytes.saturating_add(i64::try_from(command_output_bytes).unwrap_or(i64::MAX)); + output_truncated |= u64::try_from(output_bytes).unwrap_or(u64::MAX) > output_limit_bytes; if timed_out || observed_exit_code != Some(expected_exit_code) { break; } } Ok(VerifierExecution { - outcome: if !timed_out - && command_count == i64::try_from(commands.len())? - && observed_exit_code == Some(expected_exit_code) - { + outcome: if command_verifier_passed( + timed_out, + command_count, + commands.len(), + observed_exit_code, + expected_exit_code, + expected_stdout, + &captured_stdout, + stdout_bytes > stdout_capture_limit, + output_truncated, + ) { WorkflowRunVerifierOutcomeStatus::Passed } else { WorkflowRunVerifierOutcomeStatus::Failed @@ -408,6 +436,26 @@ async fn execute_command_verifier( }) } +fn command_verifier_passed( + timed_out: bool, + command_count: i64, + expected_command_count: usize, + observed_exit_code: Option, + expected_exit_code: i32, + expected_stdout: Option<&str>, + captured_stdout: &[u8], + stdout_capture_truncated: bool, + output_truncated: bool, +) -> bool { + !timed_out + && !output_truncated + && command_count == i64::try_from(expected_command_count).unwrap_or(i64::MAX) + && observed_exit_code == Some(expected_exit_code) + && expected_stdout.is_none_or(|expected| { + !stdout_capture_truncated && captured_stdout == expected.as_bytes() + }) +} + fn verifier_definition(verifier: &WorkflowRunStepVerifier) -> &Value { verifier .definition_json @@ -451,15 +499,20 @@ fn verifier_permission_profile( ) } -fn verifier_environment() -> HashMap { - ["HOME", "PATH", "LANG", "LC_ALL", "TERM"] +fn verifier_environment(workspace_cwd: &Path) -> HashMap { + let mut environment = ["PATH", "LANG", "LC_ALL", "TERM"] .into_iter() .filter_map(|key| { std::env::var(key) .ok() .map(|value| (key.to_string(), value)) }) - .collect() + .collect::>(); + environment.insert( + "HOME".to_string(), + workspace_cwd.to_string_lossy().into_owned(), + ); + environment } fn canonical_directory(path: &Path) -> anyhow::Result { @@ -491,21 +544,27 @@ fn resolve_beneath(root: &Path, relative: &str) -> anyhow::Result { async fn drain_stream( mut stream: impl AsyncRead + Unpin, - output_limit_bytes: u64, -) -> io::Result<(i64, bool)> { + capture_limit_bytes: u64, +) -> io::Result { let mut buffer = [0_u8; 8192]; let mut total = 0_u64; + let capture_capacity = usize::try_from(capture_limit_bytes) + .unwrap_or(usize::MAX) + .min(MAX_VERIFIER_CAPTURE_BYTES as usize); + let mut captured = Vec::with_capacity(capture_capacity); loop { let read = stream.read(&mut buffer).await?; if read == 0 { break; } total = total.saturating_add(u64::try_from(read).unwrap_or(u64::MAX)); + let remaining = capture_capacity.saturating_sub(captured.len()); + captured.extend_from_slice(&buffer[..read.min(remaining)]); } - Ok(( - i64::try_from(total).unwrap_or(i64::MAX), - total > output_limit_bytes, - )) + Ok(DrainedStream { + captured, + total_bytes: total, + }) } fn duration_millis(duration: Duration) -> i64 { @@ -532,4 +591,127 @@ mod tests { }); assert!(verifier_permission_profile(&definition, cwd.path()).is_err()); } + + #[tokio::test] + async fn artifact_verifier_reports_pass_and_fail() { + let workspace = tempfile::tempdir().expect("tempdir"); + std::fs::write(workspace.path().join("result.md"), "review complete") + .expect("artifact should write"); + let now = chrono::Utc::now(); + let verifier = |needle: &str| WorkflowRunStepVerifier { + verifier_run_id: "verifier-run".to_string(), + run_id: "run".to_string(), + step_id: "step".to_string(), + verifier_id: "artifact".to_string(), + verifier_type: "artifact_contains".to_string(), + status: WorkflowRunStepVerifierStatus::Running, + status_reason: None, + reason_code: None, + definition_json: serde_json::json!({ + "data": { + "artifact": "result.md", + "must_contain": [needle], + }, + }), + last_result_json: None, + attempt_count: 1, + max_attempts: None, + created_at: now, + updated_at: now, + completed_at: None, + }; + assert_eq!( + WorkflowRunVerifierOutcomeStatus::Passed, + execute_artifact_verifier(workspace.path(), &verifier("review")) + .await + .expect("matching artifact should verify") + .outcome + ); + assert_eq!( + WorkflowRunVerifierOutcomeStatus::Failed, + execute_artifact_verifier(workspace.path(), &verifier("missing")) + .await + .expect("non-matching artifact should verify") + .outcome + ); + } + + #[test] + fn command_verifier_enforces_expected_stdout() { + assert!(command_verifier_passed( + false, + 1, + 1, + Some(0), + 0, + Some(""), + b"", + false, + false, + )); + assert!(!command_verifier_passed( + false, + 1, + 1, + Some(0), + 0, + Some(""), + b"dirty\n", + false, + false, + )); + assert!(!command_verifier_passed( + false, + 1, + 1, + Some(0), + 0, + Some("expected"), + b"expected", + true, + false, + )); + assert!(!command_verifier_passed( + false, + 1, + 1, + Some(0), + 0, + None, + b"", + false, + true, + )); + } + + #[tokio::test] + async fn verifier_stream_capture_is_bounded() { + let (mut writer, reader) = tokio::io::duplex(64); + let write = tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + writer + .write_all(b"abcdef") + .await + .expect("write should work"); + }); + let drained = drain_stream(reader, 3).await.expect("stream should drain"); + write.await.expect("writer task should finish"); + assert_eq!(b"abc", drained.captured.as_slice()); + assert_eq!(6, drained.total_bytes); + } + + #[test] + fn verifier_environment_uses_workspace_home_and_whitelisted_keys() { + let workspace = tempfile::tempdir().expect("tempdir"); + let environment = verifier_environment(workspace.path()); + assert_eq!( + Some(workspace.path().to_string_lossy().as_ref()), + environment.get("HOME").map(String::as_str) + ); + assert!( + environment + .keys() + .all(|key| matches!(key.as_str(), "HOME" | "PATH" | "LANG" | "LC_ALL" | "TERM")) + ); + } } diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index 385a86e832..c5a51d8c4c 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -996,6 +996,14 @@ WHERE run_id = ? "backgroundAgentRunId": branch.background_agent_run_id.as_str(), "branchStatus": branch.status.as_str(), "reasonCode": reason_code, + "terminalReason": branch + .status_payload_json + .as_ref() + .and_then(|payload| payload.get("terminalReason")), + "workspaceCwd": branch + .status_payload_json + .as_ref() + .and_then(|payload| payload.get("cwd")), }), now_ms, }, @@ -2640,6 +2648,258 @@ WHERE plan_id = ? AND key = ? .expect("projected node should update"); } + fn passing_verifier_summary(command_count: i64) -> WorkflowRunVerifierResultSummary { + WorkflowRunVerifierResultSummary { + command_count, + expected_exit_code: (command_count > 0).then_some(0), + observed_exit_code: (command_count > 0).then_some(0), + timed_out: false, + duration_ms: 1, + output_bytes: 0, + output_truncated: false, + } + } + + #[tokio::test] + async fn workflow_start_and_successor_admission_are_exactly_once() { + let runtime = test_runtime().await; + let thread_id = test_thread_id(); + upsert_test_thread(&runtime, thread_id).await; + let marker = runtime.codex_home().join("two-step-marker"); + let (run, _) = create_projected_run( + &runtime, + thread_id, + "wf_two_step_exactly_once", + &marker, + /*include_second*/ true, + ) + .await; + let owner_id = format!("workflow-manager:{thread_id}"); + + let started = runtime + .start_workflow_run_execution(WorkflowRunStartExecutionParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: Some(2), + }) + .await + .expect("workflow should start") + .expect("workflow should be claimable"); + assert_eq!( + vec!["adversarial_scope".to_string()], + started + .admitted + .iter() + .map(|branch| branch.step_id.clone()) + .collect::>() + ); + assert_eq!( + crate::WorkflowRunStatus::Running, + started.snapshot.run.status + ); + assert!( + started.lease_expires_at_ms > datetime_to_epoch_millis(started.snapshot.run.updated_at) + ); + + let duplicate_start = runtime + .start_workflow_run_execution(WorkflowRunStartExecutionParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: Some(2), + }) + .await + .expect("duplicate start should succeed") + .expect("same owner should renew its claim"); + assert!(duplicate_start.admitted.is_empty()); + assert!(duplicate_start.generation > started.generation); + assert!(duplicate_start.lease_expires_at_ms >= started.lease_expires_at_ms); + let generation = duplicate_start.generation; + let first_branch_run_id = started.admitted[0].background_agent_run_id.clone(); + assert_eq!( + 1, + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM background_agent_runs") + .fetch_one(runtime.pool.as_ref()) + .await + .expect("background run count should load") + ); + + runtime + .update_background_agent_run_status( + first_branch_run_id.as_str(), + BackgroundAgentRunStatus::Completed, + Some("first branch completed"), + ) + .await + .expect("first branch should complete"); + let first_reconciled = runtime + .reconcile_workflow_run_branches(WorkflowRunBranchReconcileParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + }) + .await + .expect("first branch should reconcile") + .expect("workflow should remain owned"); + let first_verifier = first_reconciled + .snapshot + .verifiers + .iter() + .find(|verifier| verifier.step_id == "adversarial_scope") + .expect("first verifier should exist"); + let first_verifier_claim = runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId( + first_verifier.verifier_run_id.clone(), + ), + }) + .await + .expect("first verifier should claim") + .expect("first verifier should be ready"); + runtime + .record_workflow_run_verifier_result(WorkflowRunVerifierRecordResultParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + verifier_run_id: first_verifier_claim.verifier.verifier_run_id, + outcome: WorkflowRunVerifierOutcomeStatus::Passed, + summary: passing_verifier_summary(/*command_count*/ 1), + }) + .await + .expect("first verifier result should record") + .expect("first verifier should update"); + runtime + .advance_workflow_run(WorkflowRunAdvanceParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + }) + .await + .expect("successor should become ready") + .expect("workflow should remain owned"); + let successor = runtime + .admit_workflow_run_branches(WorkflowRunBranchAdmissionParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: Some(2), + }) + .await + .expect("successor admission should succeed") + .expect("workflow should remain owned"); + assert_eq!( + vec!["adversarial_review".to_string()], + successor + .admitted + .iter() + .map(|branch| branch.step_id.clone()) + .collect::>() + ); + let second_branch_run_id = successor.admitted[0].background_agent_run_id.clone(); + assert_ne!(first_branch_run_id, second_branch_run_id); + let duplicate_successor = runtime + .admit_workflow_run_branches(WorkflowRunBranchAdmissionParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: Some(2), + }) + .await + .expect("duplicate successor admission should succeed") + .expect("workflow should remain owned"); + assert!(duplicate_successor.admitted.is_empty()); + assert_eq!( + 2, + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM background_agent_runs") + .fetch_one(runtime.pool.as_ref()) + .await + .expect("background run count should load") + ); + + runtime + .update_background_agent_run_status( + second_branch_run_id.as_str(), + BackgroundAgentRunStatus::Completed, + Some("second branch completed"), + ) + .await + .expect("second branch should complete"); + let second_reconciled = runtime + .reconcile_workflow_run_branches(WorkflowRunBranchReconcileParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + }) + .await + .expect("second branch should reconcile") + .expect("workflow should remain owned"); + let second_verifier = second_reconciled + .snapshot + .verifiers + .iter() + .find(|verifier| verifier.step_id == "adversarial_review") + .expect("second verifier should exist"); + let second_verifier_claim = runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run.run.run_id.clone(), + owner_id: owner_id.clone(), + generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId( + second_verifier.verifier_run_id.clone(), + ), + }) + .await + .expect("second verifier should claim") + .expect("second verifier should be ready"); + let completed = runtime + .record_workflow_run_verifier_result(WorkflowRunVerifierRecordResultParams { + run_id: run.run.run_id, + owner_id, + generation, + verifier_run_id: second_verifier_claim.verifier.verifier_run_id, + outcome: WorkflowRunVerifierOutcomeStatus::Passed, + summary: passing_verifier_summary(/*command_count*/ 0), + }) + .await + .expect("second verifier result should record") + .expect("second verifier should update"); + assert_eq!( + crate::WorkflowRunStatus::Completed, + completed.snapshot.run.status + ); + assert_eq!( + 2, + completed + .snapshot + .steps + .iter() + .filter(|step| step.status == crate::WorkflowRunStepStatus::Succeeded) + .count() + ); + assert!( + !marker.exists(), + "state orchestration must not run commands" + ); + } + #[tokio::test] async fn workflow_claim_fences_stale_owner_generation() { let runtime = test_runtime().await; @@ -2704,7 +2964,7 @@ WHERE plan_id = ? AND key = ? } #[tokio::test] - async fn workflow_advance_blocks_verifiers_without_executing_commands() { + async fn workflow_advance_readies_verifiers_without_executing_commands() { let runtime = test_runtime().await; let thread_id = test_thread_id(); upsert_test_thread(&runtime, thread_id).await; @@ -2776,6 +3036,16 @@ WHERE plan_id = ? AND key = ? crate::WorkflowRunStepVerifierStatus::Pending, scope_verifier.status ); + assert_eq!( + Some(VERIFIER_READY_REASON_CODE), + advanced.snapshot.run.reason_code.as_deref() + ); + assert!(!advanced.snapshot.events.iter().any(|event| { + event + .event_payload_json + .to_string() + .contains("verifier_executor_pending") + })); assert!( !marker.exists(), "orchestrator skeleton must not execute verifier commands" @@ -3584,20 +3854,17 @@ WHERE run_id = ? let runtime = test_runtime().await; let thread_id = test_thread_id(); upsert_test_thread(&runtime, thread_id).await; - let run = create_unprojected_run( - &runtime, - thread_id, + let workflow_yaml = parallel_branch_workflow_yaml( "wf_branch_complete", - parallel_branch_workflow_yaml( - "wf_branch_complete", - /*step_count*/ 2, - /*max_parallel_steps*/ 1, - /*max_agents*/ 2, - /*max_worktrees*/ 1, - "no-secret", - ), + /*step_count*/ 2, + /*max_parallel_steps*/ 1, + /*max_agents*/ 2, + /*max_worktrees*/ 1, + "no-secret", ) - .await; + .replace(" required: []", " required:\n - \"branch-0.md\""); + let run = + create_unprojected_run(&runtime, thread_id, "wf_branch_complete", workflow_yaml).await; let claim = runtime .claim_workflow_run(WorkflowRunClaimParams { run_id: run.run.run_id.clone(), @@ -3640,6 +3907,22 @@ WHERE run_id = ? ) .await .expect("branch status should update"); + runtime + .upsert_background_agent_status_snapshot(&crate::BackgroundAgentStatusSnapshotParams { + run_id: admitted.admitted[0].background_agent_run_id.clone(), + seq: 2, + status: BackgroundAgentRunStatus::Completed, + desired_state: crate::BackgroundAgentDesiredState::Running, + summary: Some("completed".to_string()), + pending_interaction_count: 0, + last_event_seq: 1, + payload_json: json!({ + "cwd": "/tmp/workflow-branch", + "finalResult": "terminal result recorded", + }), + }) + .await + .expect("terminal status evidence should persist"); let reconciled = runtime .reconcile_workflow_run_branches(WorkflowRunBranchReconcileParams { @@ -3660,6 +3943,133 @@ WHERE run_id = ? crate::WorkflowRunStepVerifierStatus::Pending, reconciled.snapshot.verifiers[0].status ); + assert_eq!( + "branch-0.md", + reconciled.snapshot.run.artifacts_json["data"]["required"][0] + ); + let completion_event = reconciled + .snapshot + .events + .iter() + .find(|event| event.event_type == "branch_completed") + .expect("branch completion evidence should persist"); + assert_eq!( + "terminal result recorded", + completion_event.event_payload_json["data"]["terminalResult"] + ); + assert_eq!( + "/tmp/workflow-branch", + completion_event.event_payload_json["data"]["workspaceCwd"] + ); + } + + #[tokio::test] + async fn workflow_branch_failure_persists_terminal_evidence() { + let runtime = test_runtime().await; + let thread_id = test_thread_id(); + upsert_test_thread(&runtime, thread_id).await; + let run = create_unprojected_run( + &runtime, + thread_id, + "wf_branch_failure_evidence", + parallel_branch_workflow_yaml( + "wf_branch_failure_evidence", + /*step_count*/ 1, + /*max_parallel_steps*/ 1, + /*max_agents*/ 1, + /*max_worktrees*/ 1, + "no-secret", + ), + ) + .await; + let claim = runtime + .claim_workflow_run(WorkflowRunClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "owner".to_string(), + lease_duration_ms: Some(60_000), + }) + .await + .expect("claim should succeed") + .expect("run should claim"); + runtime + .advance_workflow_run(WorkflowRunAdvanceParams { + run_id: run.run.run_id.clone(), + owner_id: "owner".to_string(), + generation: claim.generation, + }) + .await + .expect("advance should succeed") + .expect("run should advance"); + let admitted = admit_test_workflow_run_branches( + &runtime, + WorkflowRunBranchAdmissionParams { + run_id: run.run.run_id.clone(), + owner_id: "owner".to_string(), + generation: claim.generation, + auth_profile_ref: None, + config_fingerprint: None, + version_fingerprint: None, + parent_agent_run_id: None, + max_active_background_agent_runs: Some(10), + }, + ) + .await + .expect("admission should succeed") + .expect("run should be owned"); + let branch_run_id = admitted.admitted[0].background_agent_run_id.clone(); + runtime + .update_background_agent_run_status( + branch_run_id.as_str(), + BackgroundAgentRunStatus::Failed, + Some("test failed"), + ) + .await + .expect("branch status should update"); + runtime + .upsert_background_agent_status_snapshot(&crate::BackgroundAgentStatusSnapshotParams { + run_id: branch_run_id, + seq: 2, + status: BackgroundAgentRunStatus::Failed, + desired_state: crate::BackgroundAgentDesiredState::Running, + summary: Some("failed".to_string()), + pending_interaction_count: 0, + last_event_seq: 1, + payload_json: json!({ + "cwd": "/tmp/workflow-failed-branch", + "terminalReason": "terminal failure recorded", + }), + }) + .await + .expect("terminal failure evidence should persist"); + + let reconciled = runtime + .reconcile_workflow_run_branches(WorkflowRunBranchReconcileParams { + run_id: run.run.run_id, + owner_id: "owner".to_string(), + generation: claim.generation, + }) + .await + .expect("reconcile should succeed") + .expect("run should be owned"); + + assert_eq!( + crate::WorkflowRunStatus::Failed, + reconciled.snapshot.run.status + ); + let failure_event = reconciled + .snapshot + .events + .iter() + .find(|event| event.event_type == "branch_failed") + .expect("branch failure evidence should persist"); + assert_eq!( + "terminal failure recorded", + failure_event.event_payload_json["data"]["terminalReason"] + ); + assert_eq!( + "/tmp/workflow-failed-branch", + failure_event.event_payload_json["data"]["workspaceCwd"] + ); } #[tokio::test] From e2fb721923185996c87186eaba14f15ec706b4f0 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 05:59:26 +0300 Subject: [PATCH 06/11] fix(workflows): stream verifier output matching Agent: vespasian --- .../thread_workflow_runtime.rs | 80 ++++++++++--------- 1 file changed, 43 insertions(+), 37 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs index 8debd0cabb..38c861ec7c 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs @@ -30,7 +30,6 @@ use tracing::warn; const WORKFLOW_RUNTIME_RECONCILE_INTERVAL: Duration = Duration::from_secs(5); const WORKFLOW_RUNTIME_RUN_LIMIT: u32 = 200; -const MAX_VERIFIER_CAPTURE_BYTES: u64 = 1024 * 1024; #[derive(Clone)] struct WorkflowRuntimeContext { @@ -45,8 +44,8 @@ struct VerifierExecution { } struct DrainedStream { - captured: Vec, total_bytes: u64, + expected_stdout_matches: bool, } impl ThreadRequestProcessor { @@ -347,10 +346,9 @@ async fn execute_command_verifier( let mut output_truncated = false; let mut timed_out = false; let mut stdout_bytes = 0_u64; - let mut captured_stdout = Vec::new(); - let stdout_capture_limit = expected_stdout - .map(|_| output_limit_bytes.min(MAX_VERIFIER_CAPTURE_BYTES)) - .unwrap_or_default(); + let mut stdout_matches = true; + let expected_stdout_bytes = + expected_stdout.map(|expected| Arc::<[u8]>::from(expected.as_bytes())); for command in commands.iter().filter_map(Value::as_str) { command_count += 1; @@ -376,8 +374,14 @@ async fn execute_command_verifier( .stderr .take() .ok_or_else(|| anyhow::anyhow!("verifier stderr was not piped"))?; - let stdout_task = tokio::spawn(drain_stream(stdout, stdout_capture_limit)); - let stderr_task = tokio::spawn(drain_stream(stderr, /* capture_limit_bytes */ 0)); + let stdout_task = tokio::spawn(drain_stream( + stdout, + expected_stdout_bytes.clone(), + stdout_bytes, + )); + let stderr_task = tokio::spawn(drain_stream( + stderr, /* expected_stdout */ None, /* expected_offset */ 0, + )); match tokio::time::timeout_at(deadline, child.wait()).await { Ok(status) => { observed_exit_code = status?.code(); @@ -390,13 +394,8 @@ async fn execute_command_verifier( } let stdout_result = stdout_task.await??; let stderr_result = stderr_task.await??; + stdout_matches &= stdout_result.expected_stdout_matches; stdout_bytes = stdout_bytes.saturating_add(stdout_result.total_bytes); - let remaining_capture = stdout_capture_limit - .saturating_sub(u64::try_from(captured_stdout.len()).unwrap_or(u64::MAX)); - let append_len = usize::try_from(remaining_capture) - .unwrap_or(usize::MAX) - .min(stdout_result.captured.len()); - captured_stdout.extend_from_slice(&stdout_result.captured[..append_len]); let command_output_bytes = stdout_result .total_bytes .saturating_add(stderr_result.total_bytes); @@ -416,8 +415,8 @@ async fn execute_command_verifier( observed_exit_code, expected_exit_code, expected_stdout, - &captured_stdout, - stdout_bytes > stdout_capture_limit, + stdout_matches, + stdout_bytes, output_truncated, ) { WorkflowRunVerifierOutcomeStatus::Passed @@ -443,8 +442,8 @@ fn command_verifier_passed( observed_exit_code: Option, expected_exit_code: i32, expected_stdout: Option<&str>, - captured_stdout: &[u8], - stdout_capture_truncated: bool, + stdout_matches: bool, + stdout_bytes: u64, output_truncated: bool, ) -> bool { !timed_out @@ -452,7 +451,7 @@ fn command_verifier_passed( && command_count == i64::try_from(expected_command_count).unwrap_or(i64::MAX) && observed_exit_code == Some(expected_exit_code) && expected_stdout.is_none_or(|expected| { - !stdout_capture_truncated && captured_stdout == expected.as_bytes() + stdout_matches && stdout_bytes == u64::try_from(expected.len()).unwrap_or(u64::MAX) }) } @@ -544,26 +543,31 @@ fn resolve_beneath(root: &Path, relative: &str) -> anyhow::Result { async fn drain_stream( mut stream: impl AsyncRead + Unpin, - capture_limit_bytes: u64, + expected_stdout: Option>, + expected_offset: u64, ) -> io::Result { let mut buffer = [0_u8; 8192]; let mut total = 0_u64; - let capture_capacity = usize::try_from(capture_limit_bytes) - .unwrap_or(usize::MAX) - .min(MAX_VERIFIER_CAPTURE_BYTES as usize); - let mut captured = Vec::with_capacity(capture_capacity); + let mut expected_stdout_matches = true; loop { let read = stream.read(&mut buffer).await?; if read == 0 { break; } + if expected_stdout_matches && let Some(expected_stdout) = expected_stdout.as_deref() { + let start = expected_offset.saturating_add(total); + let end = start.saturating_add(u64::try_from(read).unwrap_or(u64::MAX)); + expected_stdout_matches = usize::try_from(start) + .ok() + .zip(usize::try_from(end).ok()) + .and_then(|(start, end)| expected_stdout.get(start..end)) + .is_some_and(|expected| expected == &buffer[..read]); + } total = total.saturating_add(u64::try_from(read).unwrap_or(u64::MAX)); - let remaining = capture_capacity.saturating_sub(captured.len()); - captured.extend_from_slice(&buffer[..read.min(remaining)]); } Ok(DrainedStream { - captured, total_bytes: total, + expected_stdout_matches, }) } @@ -645,8 +649,8 @@ mod tests { Some(0), 0, Some(""), - b"", - false, + true, + 0, false, )); assert!(!command_verifier_passed( @@ -656,8 +660,8 @@ mod tests { Some(0), 0, Some(""), - b"dirty\n", false, + 6, false, )); assert!(!command_verifier_passed( @@ -667,8 +671,8 @@ mod tests { Some(0), 0, Some("expected"), - b"expected", - true, + false, + 8, false, )); assert!(!command_verifier_passed( @@ -678,14 +682,14 @@ mod tests { Some(0), 0, None, - b"", - false, + true, + 0, true, )); } #[tokio::test] - async fn verifier_stream_capture_is_bounded() { + async fn verifier_stream_compares_without_retaining_output() { let (mut writer, reader) = tokio::io::duplex(64); let write = tokio::spawn(async move { use tokio::io::AsyncWriteExt; @@ -694,9 +698,11 @@ mod tests { .await .expect("write should work"); }); - let drained = drain_stream(reader, 3).await.expect("stream should drain"); + let drained = drain_stream(reader, Some(Arc::from(&b"abc"[..])), 0) + .await + .expect("stream should drain"); write.await.expect("writer task should finish"); - assert_eq!(b"abc", drained.captured.as_slice()); + assert!(!drained.expected_stdout_matches); assert_eq!(6, drained.total_bytes); } From b2fb86bae7398a98e3d8fd2bde9d938b19f0d074 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 06:01:30 +0300 Subject: [PATCH 07/11] fix(workflows): isolate verifier shell startup Agent: vespasian --- .../src/request_processors/thread_workflow_runtime.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs index 38c861ec7c..a91a19f45b 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs @@ -355,7 +355,7 @@ async fn execute_command_verifier( let mut child = codex_core::exec::spawn_streaming_command_under_sandbox( vec![ "/bin/bash".to_string(), - "-lc".to_string(), + "-c".to_string(), command.to_string(), ], absolute_cwd.clone(), From 60e8dc8e5e2ff8da591fbb08e968251a3d5db77f Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 06:03:42 +0300 Subject: [PATCH 08/11] test(workflows): use valid adversarial failure fixture Agent: vespasian --- codex-rs/state/src/runtime/workflow_orchestrator.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index c5a51d8c4c..6181710bdd 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -3974,9 +3974,9 @@ WHERE run_id = ? "wf_branch_failure_evidence", parallel_branch_workflow_yaml( "wf_branch_failure_evidence", - /*step_count*/ 1, + /*step_count*/ 2, /*max_parallel_steps*/ 1, - /*max_agents*/ 1, + /*max_agents*/ 2, /*max_worktrees*/ 1, "no-secret", ), From ced44ce9be6f1bc097d7f5bb8b1ff3a2e69bd176 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 06:29:45 +0300 Subject: [PATCH 09/11] test(workflows): recover interrupted verifiers Agent: vespasian --- .../state/src/runtime/workflow_verifiers.rs | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/codex-rs/state/src/runtime/workflow_verifiers.rs b/codex-rs/state/src/runtime/workflow_verifiers.rs index a9a7e3d149..cfd2dbb568 100644 --- a/codex-rs/state/src/runtime/workflow_verifiers.rs +++ b/codex-rs/state/src/runtime/workflow_verifiers.rs @@ -1189,6 +1189,80 @@ artifacts:"#, assert!(!event_payloads.contains("true ")); } + #[tokio::test] + async fn stale_running_verifier_is_requeued_after_workflow_lease_takeover() { + let runtime = test_runtime().await; + let thread_id = test_thread_id(); + upsert_test_thread(&runtime, thread_id).await; + let (run, generation) = + create_claimed_waiting_run(&runtime, thread_id, "wf_verifier_lease_takeover", "true") + .await; + + let first_claim = runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + generation, + selection: WorkflowRunVerifierClaimSelection::NextRunCommands, + }) + .await + .expect("verifier claim should succeed") + .expect("verifier should claim"); + assert_eq!( + crate::WorkflowRunStepVerifierStatus::Running, + first_claim.verifier.status + ); + + sqlx::query("UPDATE workflow_runs SET lease_expires_at_ms = 0 WHERE run_id = ?") + .bind(run.run.run_id.as_str()) + .execute(runtime.pool.as_ref()) + .await + .expect("workflow lease should expire"); + let takeover = runtime + .claim_workflow_run(WorkflowRunClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "replacement-owner".to_string(), + lease_duration_ms: Some(60_000), + }) + .await + .expect("replacement owner claim should succeed") + .expect("expired workflow lease should be claimable"); + assert!(takeover.generation > generation); + + let recovered = takeover + .snapshot + .verifiers + .iter() + .find(|verifier| verifier.verifier_run_id == first_claim.verifier.verifier_run_id) + .expect("recovered verifier should remain in the snapshot"); + assert_eq!( + crate::WorkflowRunStepVerifierStatus::Blocked, + recovered.status + ); + assert_eq!( + Some(VERIFIER_RETRY_PENDING_REASON_CODE), + recovered.reason_code.as_deref() + ); + + let replacement_claim = runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "replacement-owner".to_string(), + generation: takeover.generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId( + recovered.verifier_run_id.clone(), + ), + }) + .await + .expect("replacement verifier claim should succeed") + .expect("interrupted verifier should be claimable again"); + assert_eq!( + crate::WorkflowRunStepVerifierStatus::Running, + replacement_claim.verifier.status + ); + assert_eq!(2, replacement_claim.verifier.attempt_count); + } + #[tokio::test] async fn verifier_pass_arms_triggered_timers_for_succeeded_step() { let runtime = test_runtime().await; From 75c697f6fb5fcb006bb22b7b38641ad09c3a645a Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 07:07:32 +0300 Subject: [PATCH 10/11] fix(workflows): execute admitted branches safely Agent: vespasian --- .../background_agent_live.rs | 6 +- .../thread_workflow_processor.rs | 4 +- .../thread_workflow_runtime.rs | 4 +- .../app-server/tests/suite/v2/workflow.rs | 139 ++++++++++++ codex-rs/background-agent/src/lib.rs | 8 +- codex-rs/ext/workflows/src/manager_tool.rs | 4 +- codex-rs/state/src/lib.rs | 7 + .../src/runtime/workflow_orchestrator.rs | 194 ++++++++++++++++- .../state/src/runtime/workflow_verifiers.rs | 200 ++++++++++++++++++ 9 files changed, 546 insertions(+), 20 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/background_agent_live.rs b/codex-rs/app-server/src/request_processors/background_agent_live.rs index 693fee2bba..0a2f0817c2 100644 --- a/codex-rs/app-server/src/request_processors/background_agent_live.rs +++ b/codex-rs/app-server/src/request_processors/background_agent_live.rs @@ -3731,7 +3731,11 @@ async fn resolve_background_agent_config( .map(serde_json::from_value::) .transpose()?; let default_permissions = payload - .and_then(|payload| payload.get("permissionProfile")) + .and_then(|payload| { + payload + .get("defaultPermissions") + .or_else(|| payload.get("permissionProfile")) + }) .and_then(Value::as_str) .map(str::to_string); let sandbox_mode = if permission_profile.is_none() { diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs b/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs index d7337a9b63..2be5ec8bc2 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_processor.rs @@ -323,7 +323,9 @@ impl ThreadWorkflowRequestProcessor { owner_id: format!("workflow-manager:{thread_id}"), auth_profile_ref: None, config_fingerprint: None, - version_fingerprint: None, + version_fingerprint: Some( + codex_state::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION.to_string(), + ), parent_agent_run_id: None, max_active_background_agent_runs: None, }) diff --git a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs index a91a19f45b..ab9a464099 100644 --- a/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_workflow_runtime.rs @@ -190,7 +190,9 @@ async fn reconcile_workflow_run( generation, auth_profile_ref: None, config_fingerprint: None, - version_fingerprint: None, + version_fingerprint: Some( + codex_state::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION.to_string(), + ), parent_agent_run_id: None, max_active_background_agent_runs: None, }) diff --git a/codex-rs/app-server/tests/suite/v2/workflow.rs b/codex-rs/app-server/tests/suite/v2/workflow.rs index f56cd31d03..077475f1ba 100644 --- a/codex-rs/app-server/tests/suite/v2/workflow.rs +++ b/codex-rs/app-server/tests/suite/v2/workflow.rs @@ -1,6 +1,7 @@ use anyhow::Result; use app_test_support::DEFAULT_CLIENT_NAME; use app_test_support::TestAppServer; +use app_test_support::create_final_assistant_message_sse_response; use app_test_support::create_fake_rollout; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; @@ -462,6 +463,86 @@ async fn workflow_run_lifecycle_projects_tasks_and_returns_sanitized_state() -> Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn started_workflow_worker_leaves_queue_and_reaches_run_commands_verifier() -> Result<()> { + let server = create_mock_responses_server_sequence_unchecked(vec![ + create_final_assistant_message_sse_response("workflow worker done")?, + ]) + .await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri(), WorkflowsFeature::Enabled)?; + let thread_id = create_materialized_thread(codex_home.path(), "workflow executor admission")?; + let yaml = executable_workflow_yaml("wf_app_server_executor_admission"); + + let mut mcp = TestAppServer::new_without_managed_config(codex_home.path()).await?; + initialize(&mut mcp, ExperimentalApiCapability::Enabled).await?; + + let create_id = send_workflow_create(&mut mcp, thread_id.as_str(), yaml.as_str()).await?; + let create_resp = read_response(&mut mcp, create_id).await?; + let ThreadWorkflowCreateResponse { workflow } = + to_response::(create_resp)?; + let start_id = mcp + .send_raw_request( + "thread/workflow/run/start", + Some(json!({ + "threadId": thread_id.as_str(), + "workflowRecordId": workflow.workflow_record_id.as_str(), + "idempotencyKey": "executor-admission", + })), + ) + .await?; + let start_resp = read_response(&mut mcp, start_id).await?; + let started = to_response::(start_resp)?; + let run_id = started.run.run.run_id; + let runtime = open_state_runtime(codex_home.path()).await?; + + let (completed, worker) = timeout(std::time::Duration::from_secs(30), async { + loop { + let snapshot = runtime + .workflows() + .get_workflow_run_snapshot(run_id.as_str()) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow run disappeared"))?; + if let Some(background_agent_run_id) = snapshot + .steps + .first() + .and_then(|step| step.background_agent_run_id.as_deref()) + && let Some(worker) = runtime + .get_background_agent_run(background_agent_run_id) + .await? + && worker.status == codex_state::BackgroundAgentRunStatus::Completed + && snapshot.run.status == codex_state::WorkflowRunStatus::Completed + && snapshot.verifiers.iter().any(|verifier| { + verifier.status == codex_state::WorkflowRunStepVerifierStatus::Passed + }) + { + return Ok::<_, anyhow::Error>((snapshot, worker)); + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + }) + .await??; + + assert_eq!( + Some(codex_state::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION), + worker.version_fingerprint.as_deref() + ); + assert!( + completed + .events + .iter() + .any(|event| event.event_type == "verifier_started") + ); + assert!( + completed + .events + .iter() + .any(|event| event.event_type == "verifier_passed") + ); + + Ok(()) +} + #[tokio::test] async fn workflow_create_returns_sanitized_validation_error() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; @@ -801,6 +882,64 @@ cleanup: ) } +fn executable_workflow_yaml(workflow_id: &str) -> String { + format!( + r#"schema_version: "workflow.codex.codewith/v0" +workflow_id: "{workflow_id}" +display_name: "Workflow Executor Admission" +source_prompt: "Run one real workflow branch and its deterministic verifier." +status: "draft" +execution_defaults: + model_gateway: "hasna" + provider: "mock_provider" + model: "mock-model" + reasoning: "high" +limits: + max_parallel_steps: 1 + max_agents: 1 + max_worktrees: 1 + max_runtime_seconds: 60 + max_step_runtime_seconds: 30 + max_tokens: 1000 + max_tool_calls: 10 +approvals: + required_before: [] +agents: + - id: "executor" + display_name: "Executor-Archimedes" + role: "Complete the workflow branch." + model: + model_gateway: "hasna" + provider: "mock_provider" + model: "mock-model" + reasoning: "high" +steps: + - id: "execute" + title: "Execute the admitted branch" + agent: "executor" + depends_on: [] + completion: + model_marked_state: "candidate_succeeded" + verifiers: + - id: "command_check" + type: "run_commands" + cwd: "." + sandbox: "read-only" + network: "disabled" + timeout_seconds: 5 + output_limit_bytes: 1024 + commands: + - "true" +artifacts: + retention: "preserve_evidence" + required: [] +cleanup: + on_cancel: [] + on_complete: [] +"# + ) +} + fn yaml_single_quoted(value: &str) -> String { format!("'{}'", value.replace('\'', "''")) } diff --git a/codex-rs/background-agent/src/lib.rs b/codex-rs/background-agent/src/lib.rs index 6494ffae99..2d30620d81 100644 --- a/codex-rs/background-agent/src/lib.rs +++ b/codex-rs/background-agent/src/lib.rs @@ -6,7 +6,9 @@ pub mod process_lifecycle; mod supervisor; pub mod worker_admission; +pub use codex_state::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION; pub use codex_state::BACKGROUND_AGENT_EVENT_CURSOR_COMPACTED; +pub use codex_state::BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT; pub use codex_state::BackgroundAgentDesiredState; pub use codex_state::BackgroundAgentEvent; pub use codex_state::BackgroundAgentExecutionHandleParams; @@ -30,8 +32,6 @@ pub use codex_state::BackgroundAgentWorktreeLeaseCreateParams; pub use supervisor::DurableAgentSupervisor; pub use supervisor::DurableAgentSupervisorConfig; -pub const BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION: &str = - "codewith.background-agent.admission.v1"; pub const BACKGROUND_AGENT_ADMISSION_CAPACITY_EXCEEDED: &str = "background_agent_admission_capacity_exceeded"; pub const BACKGROUND_AGENT_ADMISSION_IDENTITY_MISMATCH: &str = @@ -42,10 +42,6 @@ pub const BACKGROUND_AGENT_ADMISSION_SCHEMA_MISMATCH: &str = "background_agent_admission_schema_mismatch"; pub const BACKGROUND_AGENT_DAEMON_INCOMPATIBLE: &str = "background_agent_daemon_incompatible"; pub const BACKGROUND_AGENT_DAEMON_PROTOCOL_VERSION: u32 = 1; -pub const BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT: &str = concat!( - "codewith.background-agent.runtime.v1:", - env!("CARGO_PKG_VERSION") -); pub const DEFAULT_MAX_ACTIVE_BACKGROUND_AGENT_RUNS: i64 = 8; /// Durable run roster used by the background-agent supervisor. diff --git a/codex-rs/ext/workflows/src/manager_tool.rs b/codex-rs/ext/workflows/src/manager_tool.rs index 99b1602722..5335c4e8b2 100644 --- a/codex-rs/ext/workflows/src/manager_tool.rs +++ b/codex-rs/ext/workflows/src/manager_tool.rs @@ -342,7 +342,9 @@ impl ManageWorkflowTool { owner_id: format!("workflow-manager:{thread_id}"), auth_profile_ref: None, config_fingerprint: None, - version_fingerprint: None, + version_fingerprint: Some( + codex_state::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION.to_string(), + ), parent_agent_run_id: None, max_active_background_agent_runs: None, }) diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 037b050726..1eafe71bad 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -25,6 +25,13 @@ pub use runtime::StateRuntimeStartupLock; pub use runtime::acquire_state_runtime_startup_lock; pub use runtime::state_runtime_startup_lock_path; +pub const BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION: &str = + "codewith.background-agent.admission.v1"; +pub const BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT: &str = concat!( + "codewith.background-agent.runtime.v1:", + env!("CARGO_PKG_VERSION") +); + pub use audit::ThreadStateAuditRow; pub use audit::read_thread_state_audit_rows; /// Low-level storage engine: useful for focused tests. diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index 6181710bdd..e9337339ba 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -8,6 +8,7 @@ use crate::runtime::background_agents::insert_background_agent_run_in_tx; use crate::runtime::background_agents::recover_or_validate_background_agent_initial_state_in_tx; use crate::runtime::background_agents::validate_existing_background_agent_admission_in_tx; use crate::runtime::workflow_automation::arm_workflow_timers_for_succeeded_step_in_tx; +use crate::runtime::workflow_verifiers::requeue_running_workflow_verifiers_in_tx; use crate::runtime::workflows::WorkflowRunEventAppend; use crate::runtime::workflows::append_workflow_run_event_in_tx; use crate::runtime::workflows::maybe_snapshot_workflow_run_in_tx; @@ -31,6 +32,7 @@ const WORKFLOW_BRANCH_PROVIDER_ENV_MISSING_REASON: &str = "OpenRouter workflow branch requires OPENROUTER_API_KEY before admission"; const WORKFLOW_BRANCH_PROVIDER_ENV_MISSING_REASON_CODE: &str = "workflow_branch_provider_env_missing"; +const WORKFLOW_BRANCH_RECOVERY_POLICY: &str = "abort_mid_turn_resume_at_safe_boundary"; const WORKFLOW_BRANCH_SOURCE: &str = "workflow"; const WORKFLOW_BRANCH_THREAD_STORE_KIND: &str = "background-agent"; @@ -340,7 +342,7 @@ RETURNING generation WorkflowRunEventAppend { event_type: "claimed", actor_kind: "orchestrator", - actor_id: Some(params.owner_id), + actor_id: Some(params.owner_id.clone()), step_run_id: None, verifier_run_id: None, visibility: "internal", @@ -352,6 +354,14 @@ RETURNING generation }, ) .await?; + requeue_running_workflow_verifiers_in_tx( + &mut tx, + params.run_id.as_str(), + params.owner_id.as_str(), + generation, + now_ms, + ) + .await?; let snapshot = snapshot_workflow_run_in_tx(&mut tx, params.run_id.as_str()).await?; tx.commit().await?; Ok(Some(WorkflowRunClaimOutcome { @@ -1077,6 +1087,13 @@ struct BackgroundBranchRunCreate<'a> { now_ms: i64, } +struct WorkflowBranchExecutionIdentity { + cwd: String, + workspace_roots: Vec, + config_fingerprint: String, + version_fingerprint: String, +} + struct BackgroundAgentStatusSnapshotUpsert<'a> { run_id: &'a str, seq: i64, @@ -1314,6 +1331,14 @@ async fn create_background_branch_run_if_missing_in_tx( agent_id: candidate.agent_id.as_str(), parallel_group: candidate.parallel_group.as_deref(), }); + let prompt_sha256 = StateRuntime::background_agent_identity_sha256(prompt.as_bytes()); + let execution_identity = workflow_branch_execution_identity( + run, + candidate, + model_route_json, + workspace_json, + params, + )?; let prompt_snapshot_ref = format!("workflow:{}:step:{}:prompt", run.run_id, candidate.step_id); let spawn_linkage_json = json!({ "schemaVersion": "workflow.branch_spawn/v0", @@ -1345,13 +1370,15 @@ async fn create_background_branch_run_if_missing_in_tx( spawn_linkage_json: Some(spawn_linkage_json), auth_profile_ref: params.auth_profile_ref.clone(), status_reason: Some("queued by workflow branch admission".to_string()), - config_fingerprint: params.config_fingerprint.clone(), - version_fingerprint: params.version_fingerprint.clone(), + config_fingerprint: Some(execution_identity.config_fingerprint.clone()), + version_fingerprint: Some(execution_identity.version_fingerprint.clone()), }; let start_event_payload = json!({ - "cwd": Value::Null, + "cwd": execution_identity.cwd.as_str(), "prompt": prompt, + "promptSha256": prompt_sha256, "promptSnapshotRef": prompt_snapshot_ref, + "initialGoalObjective": Value::Null, }); let execution_snapshot_params = BackgroundAgentExecutionSnapshotParams { run_id: background_agent_run_id.to_string(), @@ -1362,9 +1389,10 @@ async fn create_background_branch_run_if_missing_in_tx( model_route_json, workspace_json, params, + &execution_identity, ), - recovery_policy: "abort_mid_turn_resume_at_safe_boundary".to_string(), - config_fingerprint: params.config_fingerprint.clone(), + recovery_policy: WORKFLOW_BRANCH_RECOVERY_POLICY.to_string(), + config_fingerprint: Some(execution_identity.config_fingerprint.clone()), }; let admission_identity_sha256 = background_agent_admission_identity_sha256( &run_params, @@ -1465,12 +1493,75 @@ ON CONFLICT(run_id) DO UPDATE SET Ok(()) } +fn workflow_branch_execution_identity( + run: &crate::WorkflowRun, + candidate: &ReadyBranchCandidate, + model_route_json: &Value, + workspace_json: Option<&Value>, + params: &WorkflowRunBranchAdmissionParams, +) -> anyhow::Result { + let version_fingerprint = params + .version_fingerprint + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or(crate::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION); + if version_fingerprint != crate::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION { + anyhow::bail!( + "background_agent_admission_schema_mismatch: workflow branch requested incompatible background-agent admission schema `{version_fingerprint}`" + ); + } + + let cwd = std::env::current_dir() + .map_err(|err| anyhow::anyhow!("failed to resolve workflow branch cwd: {err}"))? + .to_string_lossy() + .into_owned(); + if cwd.is_empty() { + anyhow::bail!("workflow branch cwd must not be empty"); + } + let workspace_roots = vec![cwd.clone()]; + let config_fingerprint = if let Some(config_fingerprint) = params + .config_fingerprint + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + { + config_fingerprint.to_string() + } else { + let auth_profile_identity_sha256 = + params.auth_profile_ref.as_deref().map(|profile| { + StateRuntime::background_agent_identity_sha256(profile.as_bytes()) + }); + let config_identity = json!({ + "snapshotSource": "workflow/branch_admission", + "workflowRunId": run.run_id.as_str(), + "workflowStepId": candidate.step_id.as_str(), + "workflowStepRunId": candidate.step_run_id.as_str(), + "cwd": cwd.as_str(), + "workspaceRoots": &workspace_roots, + "modelRoute": model_route_json, + "workspace": workspace_json, + "authProfileIdentitySha256": auth_profile_identity_sha256, + }); + let serialized = serde_json::to_vec(&config_identity)?; + StateRuntime::background_agent_identity_sha256(serialized.as_slice()) + }; + + Ok(WorkflowBranchExecutionIdentity { + cwd, + workspace_roots, + config_fingerprint, + version_fingerprint: version_fingerprint.to_string(), + }) +} + fn branch_execution_payload( run: &crate::WorkflowRun, candidate: &ReadyBranchCandidate, model_route_json: &Value, workspace_json: Option<&Value>, params: &WorkflowRunBranchAdmissionParams, + execution_identity: &WorkflowBranchExecutionIdentity, ) -> Value { json!({ "snapshotSource": "workflow/branch_admission", @@ -1478,22 +1569,36 @@ fn branch_execution_payload( "workflowStepId": candidate.step_id.as_str(), "workflowStepRunId": candidate.step_run_id.as_str(), "agentId": candidate.agent_id.as_str(), - "cwd": Value::Null, - "workspaceRoots": Value::Null, + "cwd": execution_identity.cwd.as_str(), + "initialGoalObjective": Value::Null, + "workspaceRoots": &execution_identity.workspace_roots, "modelGateway": model_route_json.get("model_gateway"), "model": model_route_json.get("model"), "provider": model_route_json.get("provider"), "reasoning": model_route_json.get("reasoning"), "serviceTier": model_route_json.get("service_tier"), "approvalPolicy": model_route_json.get("approval_policy"), - "permissionProfile": model_route_json.get("permission_profile"), + "permissionProfile": Value::Null, + "defaultPermissions": model_route_json.get("permission_profile"), + "sandboxPolicy": Value::Null, + "networkPolicy": Value::Null, + "mcpToolAllowlist": Value::Null, "authProfileIdentitySha256": params .auth_profile_ref .as_deref() .map(|profile| StateRuntime::background_agent_identity_sha256(profile.as_bytes())), + "managedWorktreeId": Value::Null, "workspace": workspace_json, "envSnapshotPolicy": "inherit-minimal", + "shellSnapshot": Value::Null, + "configSourceHashes": Value::Null, "maxRuntimeSeconds": workflow_state_data(&run.limits_json).get("max_step_runtime_seconds"), + "maxTokens": workflow_state_data(&run.limits_json).get("max_tokens"), + "configFingerprint": execution_identity.config_fingerprint.as_str(), + "versionFingerprint": execution_identity.version_fingerprint.as_str(), + "packageFingerprint": crate::BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT, + "recoveryPolicy": WORKFLOW_BRANCH_RECOVERY_POLICY, + "midTurnCrashSemantics": WORKFLOW_BRANCH_RECOVERY_POLICY, }) } @@ -3106,7 +3211,9 @@ WHERE plan_id = ? AND key = ? generation: claim.generation, auth_profile_ref: Some("profile:workflow".to_string()), config_fingerprint: Some("cfg-workflow".to_string()), - version_fingerprint: Some("version-workflow".to_string()), + version_fingerprint: Some( + crate::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION.to_string(), + ), parent_agent_run_id: None, max_active_background_agent_runs: Some(10), }, @@ -3164,6 +3271,25 @@ WHERE plan_id = ? AND key = ? Some("profile:workflow"), first_run.auth_profile_ref.as_deref() ); + assert_eq!( + Some("cfg-workflow"), + first_run.config_fingerprint.as_deref() + ); + assert_eq!( + Some(crate::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION), + first_run.version_fingerprint.as_deref() + ); + assert!( + runtime + .background_agent_admission_is_ready( + first_branch.background_agent_run_id.as_str(), + crate::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION, + crate::BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT, + ) + .await + .expect("background run admission readiness should load"), + "workflow branch must persist a supervisor-compatible admission envelope" + ); let execution_snapshot = runtime .get_latest_background_agent_execution_snapshot( first_branch.background_agent_run_id.as_str(), @@ -3185,6 +3311,54 @@ WHERE plan_id = ? AND key = ? .get("reasoning") .and_then(Value::as_str) ); + assert!( + execution_snapshot + .payload_json + .get("cwd") + .and_then(Value::as_str) + .is_some_and(|cwd| !cwd.is_empty()) + ); + assert_eq!( + Some("cfg-workflow"), + execution_snapshot + .payload_json + .get("configFingerprint") + .and_then(Value::as_str) + ); + assert_eq!( + Some(crate::BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION), + execution_snapshot + .payload_json + .get("versionFingerprint") + .and_then(Value::as_str) + ); + assert_eq!( + Some(crate::BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT), + execution_snapshot + .payload_json + .get("packageFingerprint") + .and_then(Value::as_str) + ); + assert_eq!( + Some(WORKFLOW_BRANCH_RECOVERY_POLICY), + execution_snapshot + .payload_json + .get("recoveryPolicy") + .and_then(Value::as_str) + ); + assert!( + execution_snapshot + .payload_json + .get("permissionProfile") + .is_some_and(Value::is_null) + ); + assert_eq!( + Some("workspace-write"), + execution_snapshot + .payload_json + .get("defaultPermissions") + .and_then(Value::as_str) + ); } #[tokio::test] diff --git a/codex-rs/state/src/runtime/workflow_verifiers.rs b/codex-rs/state/src/runtime/workflow_verifiers.rs index cfd2dbb568..3510a0bdcf 100644 --- a/codex-rs/state/src/runtime/workflow_verifiers.rs +++ b/codex-rs/state/src/runtime/workflow_verifiers.rs @@ -291,6 +291,74 @@ WHERE run_id = ? } } +pub(super) async fn requeue_running_workflow_verifiers_in_tx( + tx: &mut sqlx::Transaction<'_, Sqlite>, + run_id: &str, + owner_id: &str, + generation: i64, + now_ms: i64, +) -> anyhow::Result { + let rows = sqlx::query( + r#" +UPDATE workflow_run_step_verifiers +SET + status = ?, + status_reason = ?, + reason_code = ?, + completed_at_ms = NULL, + updated_at_ms = ? +WHERE run_id = ? + AND status = 'running' +RETURNING + verifier_run_id, + step_id, + attempt_count, + ( + SELECT step_run_id + FROM workflow_run_steps step + WHERE step.run_id = workflow_run_step_verifiers.run_id + AND step.step_id = workflow_run_step_verifiers.step_id + ) AS step_run_id + "#, + ) + .bind(crate::WorkflowRunStepVerifierStatus::Blocked.as_str()) + .bind(VERIFIER_RETRY_PENDING_REASON) + .bind(VERIFIER_RETRY_PENDING_REASON_CODE) + .bind(now_ms) + .bind(run_id) + .fetch_all(&mut **tx) + .await?; + + for row in &rows { + let verifier_run_id: String = row.try_get("verifier_run_id")?; + let step_id: String = row.try_get("step_id")?; + let step_run_id: String = row.try_get("step_run_id")?; + let attempt_count: i64 = row.try_get("attempt_count")?; + append_workflow_run_event_in_tx( + tx, + run_id, + WorkflowRunEventAppend { + event_type: "verifier_interrupted", + actor_kind: "orchestrator", + actor_id: Some(owner_id.to_string()), + step_run_id: Some(step_run_id), + verifier_run_id: Some(verifier_run_id), + visibility: "internal", + payload: json!({ + "stepId": step_id, + "generation": generation, + "attempt": attempt_count, + "reasonCode": VERIFIER_RETRY_PENDING_REASON_CODE, + }), + now_ms, + }, + ) + .await?; + } + + Ok(rows.len()) +} + async fn claim_next_run_commands_verifier_in_tx( tx: &mut sqlx::Transaction<'_, Sqlite>, run_id: &str, @@ -1189,6 +1257,122 @@ artifacts:"#, assert!(!event_payloads.contains("true ")); } + #[tokio::test] + async fn running_verifier_is_requeued_when_same_owner_reclaims_after_restart() { + let runtime = test_runtime().await; + let thread_id = test_thread_id(); + upsert_test_thread(&runtime, thread_id).await; + let (run, generation) = + create_claimed_waiting_run(&runtime, thread_id, "wf_verifier_restart", "true").await; + + let first_claim = runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + generation, + selection: WorkflowRunVerifierClaimSelection::NextRunCommands, + }) + .await + .expect("verifier claim should succeed") + .expect("verifier should claim"); + assert_eq!( + crate::WorkflowRunStepVerifierStatus::Running, + first_claim.verifier.status + ); + + let restarted = runtime + .claim_workflow_run(WorkflowRunClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + lease_duration_ms: Some(60_000), + }) + .await + .expect("restarted owner claim should succeed") + .expect("same owner should reclaim the workflow"); + assert!(restarted.generation > generation); + + let recovered = restarted + .snapshot + .verifiers + .iter() + .find(|verifier| verifier.verifier_run_id == first_claim.verifier.verifier_run_id) + .expect("recovered verifier should remain in the snapshot"); + assert_eq!( + crate::WorkflowRunStepVerifierStatus::Blocked, + recovered.status + ); + assert_eq!( + Some(VERIFIER_RETRY_PENDING_REASON_CODE), + recovered.reason_code.as_deref() + ); + + let stale_result = runtime + .record_workflow_run_verifier_result(WorkflowRunVerifierRecordResultParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + generation, + verifier_run_id: recovered.verifier_run_id.clone(), + outcome: WorkflowRunVerifierOutcomeStatus::Passed, + summary: passing_summary(), + }) + .await + .expect("stale verifier result should be rejected without an error"); + assert!( + stale_result.is_none(), + "the previous generation must not complete the requeued verifier" + ); + + let replacement_claim = runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + generation: restarted.generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId( + recovered.verifier_run_id.clone(), + ), + }) + .await + .expect("replacement verifier claim should succeed") + .expect("interrupted verifier should be claimable again"); + assert_eq!( + crate::WorkflowRunStepVerifierStatus::Running, + replacement_claim.verifier.status + ); + assert_eq!(2, replacement_claim.verifier.attempt_count); + + let recorded = runtime + .record_workflow_run_verifier_result(WorkflowRunVerifierRecordResultParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + generation: restarted.generation, + verifier_run_id: replacement_claim.verifier.verifier_run_id, + outcome: WorkflowRunVerifierOutcomeStatus::Passed, + summary: passing_summary(), + }) + .await + .expect("replacement verifier result should record") + .expect("replacement verifier result should update"); + assert_eq!(crate::WorkflowRunStatus::Completed, recorded.snapshot.run.status); + assert_eq!( + 1, + recorded + .snapshot + .events + .iter() + .filter(|event| event.event_type == "verifier_passed") + .count(), + "only the replacement generation may complete the verifier" + ); + assert!( + recorded + .snapshot + .events + .iter() + .any(|event| event.event_type == "verifier_interrupted"), + "restart recovery should leave an explicit audit event" + ); + } + #[tokio::test] async fn stale_running_verifier_is_requeued_after_workflow_lease_takeover() { let runtime = test_runtime().await; @@ -1244,6 +1428,22 @@ artifacts:"#, recovered.reason_code.as_deref() ); + let stale_result = runtime + .record_workflow_run_verifier_result(WorkflowRunVerifierRecordResultParams { + run_id: run.run.run_id.clone(), + owner_id: "verifier-owner".to_string(), + generation, + verifier_run_id: recovered.verifier_run_id.clone(), + outcome: WorkflowRunVerifierOutcomeStatus::Passed, + summary: passing_summary(), + }) + .await + .expect("stale verifier result should be rejected without an error"); + assert!( + stale_result.is_none(), + "the previous owner and generation must not complete the requeued verifier" + ); + let replacement_claim = runtime .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { run_id: run.run.run_id.clone(), From 59d6ffe41b733ac0c28585d3b48129501a835e20 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sat, 8 Aug 2026 07:13:58 +0300 Subject: [PATCH 11/11] style(workflows): apply remote rustfmt Agent: vespasian --- codex-rs/app-server/tests/suite/v2/workflow.rs | 2 +- codex-rs/state/src/runtime/workflow_orchestrator.rs | 8 ++++---- codex-rs/state/src/runtime/workflow_verifiers.rs | 5 ++++- 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/workflow.rs b/codex-rs/app-server/tests/suite/v2/workflow.rs index 077475f1ba..ad56f9d7a7 100644 --- a/codex-rs/app-server/tests/suite/v2/workflow.rs +++ b/codex-rs/app-server/tests/suite/v2/workflow.rs @@ -1,8 +1,8 @@ use anyhow::Result; use app_test_support::DEFAULT_CLIENT_NAME; use app_test_support::TestAppServer; -use app_test_support::create_final_assistant_message_sse_response; use app_test_support::create_fake_rollout; +use app_test_support::create_final_assistant_message_sse_response; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; use app_test_support::write_mock_provider_models_cache; diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index e9337339ba..b507c388b3 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -1528,10 +1528,10 @@ fn workflow_branch_execution_identity( { config_fingerprint.to_string() } else { - let auth_profile_identity_sha256 = - params.auth_profile_ref.as_deref().map(|profile| { - StateRuntime::background_agent_identity_sha256(profile.as_bytes()) - }); + let auth_profile_identity_sha256 = params + .auth_profile_ref + .as_deref() + .map(|profile| StateRuntime::background_agent_identity_sha256(profile.as_bytes())); let config_identity = json!({ "snapshotSource": "workflow/branch_admission", "workflowRunId": run.run_id.as_str(), diff --git a/codex-rs/state/src/runtime/workflow_verifiers.rs b/codex-rs/state/src/runtime/workflow_verifiers.rs index 3510a0bdcf..eb639d8c42 100644 --- a/codex-rs/state/src/runtime/workflow_verifiers.rs +++ b/codex-rs/state/src/runtime/workflow_verifiers.rs @@ -1352,7 +1352,10 @@ artifacts:"#, .await .expect("replacement verifier result should record") .expect("replacement verifier result should update"); - assert_eq!(crate::WorkflowRunStatus::Completed, recorded.snapshot.run.status); + assert_eq!( + crate::WorkflowRunStatus::Completed, + recorded.snapshot.run.status + ); assert_eq!( 1, recorded