diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 31de78ca43..fb614e0acc 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4306,14 +4306,17 @@ dependencies = [ "codex-utils-absolute-path", "codex-utils-output-truncation", "codex-workflows", + "libc", "pretty_assertions", "serde", "serde_json", + "serde_yaml", "sha2 0.10.9", "tempfile", "tokio", "tokio-util", "tracing", + "windows-sys 0.52.0", ] [[package]] diff --git a/codex-rs/app-server/tests/suite/v2/workflow.rs b/codex-rs/app-server/tests/suite/v2/workflow.rs index 3be7fa3a39..24176498fa 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_apply_patch_sse_response; use app_test_support::create_fake_rollout; use app_test_support::create_fake_rollout_with_cwd; use app_test_support::create_final_assistant_message_sse_response; @@ -42,7 +43,6 @@ const DEFAULT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const ACTIVATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(45); const RAW_SENTINEL: &str = "RAW_WORKFLOW_SECRET_SHOULD_NOT_LEAK"; const COMMAND_SENTINEL: &str = "WORKFLOW_COMMAND_SHOULD_NOT_RUN_OR_LEAK"; - #[derive(Debug, Clone, Copy)] enum WorkflowsFeature { Enabled, @@ -479,24 +479,38 @@ async fn workflow_run_lifecycle_projects_tasks_and_returns_sanitized_state() -> } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn workflow_start_activates_one_real_worker_and_one_verifier() -> Result<()> { +async fn workflow_start_activates_paired_real_workers_and_verifiers() -> Result<()> { let codex_home = TempDir::new()?; let source_repo = TempDir::new()?; init_git_repo(source_repo.path())?; let source_head = git_output(source_repo.path(), &["rev-parse", "HEAD"])?; let source_status = git_output(source_repo.path(), &["status", "--short"])?; - let server = - create_mock_responses_server_sequence(vec![create_final_assistant_message_sse_response( - "workflow worker done", - )?]) - .await; - create_config_toml(codex_home.path(), &server.uri(), WorkflowsFeature::Enabled)?; + let yaml = actual_worker_workflow_yaml(); + let review_artifact = finite_review_artifact(source_head.as_str(), yaml.as_str())?; + let review_patch = format!( + "*** Begin Patch\n*** Add File: review.yaml\n{}*** End Patch\n", + review_artifact + .lines() + .map(|line| format!("+{line}\n")) + .collect::() + ); + let server = create_mock_responses_server_sequence(vec![ + create_final_assistant_message_sse_response("workflow candidate done")?, + create_apply_patch_sse_response(review_patch.as_str(), "review-artifact-patch")?, + create_final_assistant_message_sse_response("workflow review done")?, + ]) + .await; + create_config_toml_with_sandbox_mode( + codex_home.path(), + &server.uri(), + WorkflowsFeature::Enabled, + "workspace-write", + )?; let thread_id = create_materialized_thread_with_cwd( codex_home.path(), "workflow actual worker", source_repo.path(), )?; - let yaml = actual_worker_workflow_yaml(); codex_workflows::parse_workflow_yaml(&yaml)?; let mut mcp = TestAppServer::new_without_managed_config(codex_home.path()).await?; @@ -539,25 +553,75 @@ async fn workflow_start_activates_one_real_worker_and_one_verifier() -> Result<( codex_state::WorkflowRunStatus::Completed, snapshot.run.status ); - assert_eq!(1, snapshot.steps.len()); + assert_eq!(2, snapshot.steps.len()); + assert!( + snapshot + .steps + .iter() + .all(|step| step.status == codex_state::WorkflowRunStepStatus::Succeeded) + ); + let review_step = snapshot + .steps + .iter() + .find(|step| step.step_id == "initial_adversarial_review") + .ok_or_else(|| anyhow::anyhow!("workflow review step is missing"))?; + let review_admission = review_step + .branch_admission_json + .as_ref() + .and_then(|value| value.get("data")) + .ok_or_else(|| anyhow::anyhow!("workflow review admission is missing"))?; + assert_eq!( + Some(source_head.as_str()), + review_admission + .pointer("/reviewContext/candidateIdentity") + .and_then(serde_json::Value::as_str) + ); + assert_eq!( + Some(false), + review_admission + .pointer("/reviewContext/artifactPreexisting") + .and_then(serde_json::Value::as_bool) + ); + let review_managed_worktree_id = review_admission + .get("managedWorktreeId") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .ok_or_else(|| { + anyhow::anyhow!("workflow review branch admission has no managedWorktreeId") + })?; + let review_managed_worktree = runtime + .managed_worktrees() + .get_managed_worktree(review_managed_worktree_id) + .await? + .ok_or_else(|| { + anyhow::anyhow!( + "workflow review managed worktree {review_managed_worktree_id} was not persisted" + ) + })?; assert_eq!( - codex_state::WorkflowRunStepStatus::Succeeded, - snapshot.steps[0].status + codex_state::ManagedWorktreeMode::IsolatedWorktree, + review_managed_worktree.mode ); - assert_eq!(1, snapshot.verifiers.len()); assert_eq!( - codex_state::WorkflowRunStepVerifierStatus::Passed, - snapshot.verifiers[0].status + review_artifact, + std::fs::read_to_string(review_managed_worktree.worktree_path.join("review.yaml"))? + ); + assert_eq!(2, snapshot.verifiers.len()); + assert!( + snapshot + .verifiers + .iter() + .all(|verifier| verifier.status == codex_state::WorkflowRunStepVerifierStatus::Passed) ); for event_type in ["branch_admitted", "verifier_started", "verifier_passed"] { assert_eq!( - 1, + 2, snapshot .events .iter() .filter(|event| event.event_type == event_type) .count(), - "workflow should record exactly one {event_type} event" + "workflow should record exactly two {event_type} events" ); } @@ -593,7 +657,7 @@ async fn workflow_start_activates_one_real_worker_and_one_verifier() -> Result<( .and_then(|value| value.get("agentId")) .and_then(serde_json::Value::as_str) ); - assert_eq!(1, runtime.list_background_agent_runs(Some(10)).await?.len()); + assert_eq!(2, runtime.list_background_agent_runs(Some(10)).await?.len()); let managed_worktree = runtime .managed_worktrees() @@ -799,6 +863,15 @@ fn create_config_toml( codex_home: &Path, server_uri: &str, workflows_feature: WorkflowsFeature, +) -> std::io::Result<()> { + create_config_toml_with_sandbox_mode(codex_home, server_uri, workflows_feature, "read-only") +} + +fn create_config_toml_with_sandbox_mode( + codex_home: &Path, + server_uri: &str, + workflows_feature: WorkflowsFeature, + sandbox_mode: &str, ) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); std::fs::write( @@ -807,7 +880,7 @@ fn create_config_toml( r#" model = "mock-model" approval_policy = "never" -sandbox_mode = "read-only" +sandbox_mode = "{sandbox_mode}" model_provider = "mock_provider" suppress_unstable_features_warning = true @@ -860,11 +933,51 @@ fn invalid_fenced_yaml(raw_sentinel: &str, marker: &Path) -> String { ) } +fn finite_review_artifact(candidate_identity: &str, workflow_yaml: &str) -> Result { + let spec = codex_workflows::parse_workflow_yaml(workflow_yaml)?; + let verifier = spec + .steps + .iter() + .find(|step| step.id == "actual_worker") + .and_then(|step| step.completion.as_ref()) + .and_then(|completion| completion.verifiers.first()) + .ok_or_else(|| anyhow::anyhow!("actual worker verifier is missing"))?; + let definition_canonical = serde_json::to_string(verifier)?; + let outcome = json!({ + "status": "passed", + "expectedExitCode": 0, + "observedExitCode": 0, + "timedOut": false, + "outputTruncated": false, + }); + let criteria = [ + format!( + "dependency `actual_worker` verifier `worktree_git_status` (run_commands) contract: {definition_canonical}" + ), + format!( + "dependency `actual_worker` verifier `worktree_git_status` observed outcome: {}", + serde_json::to_string(&outcome)? + ), + format!( + "dependency `actual_worker` completed as `succeeded` at exact HEAD `{candidate_identity}`" + ), + ]; + let criteria = criteria + .iter() + .map(|criterion| { + serde_json::to_string(criterion).map(|criterion| format!(" - {criterion}\n")) + }) + .collect::>()?; + Ok(format!( + "candidate_identity: {candidate_identity}\nacceptance_criteria:\n{criteria}verdict: GO\nblocking_p0_p1: []\nnon_blocking_p2_p3: []\nremediation_cycle: 0\nremediation_cycle_cap: 2\n" + )) +} + fn actual_worker_workflow_yaml() -> String { r#"schema_version: "workflow.codex.codewith/v0" workflow_id: "wf_app_server_actual_worker" display_name: "Actual Worker Activation" -source_prompt: "Run one real adversarial worker and verify its isolated checkout." +source_prompt: "Run one real candidate worker, then one independent adversarial reviewer." status: "draft" execution_defaults: model_gateway: "hasna" @@ -875,8 +988,8 @@ execution_defaults: permission_profile: "read-only" limits: max_parallel_steps: 1 - max_agents: 1 - max_worktrees: 1 + max_agents: 2 + max_worktrees: 2 max_runtime_seconds: 120 max_step_runtime_seconds: 60 max_tokens: 10000 @@ -884,9 +997,9 @@ limits: approvals: required_before: [] agents: - - id: "adversarial_worker" - display_name: "Adversary-Hypatia" - role: "Adversarially verify actual workflow worker activation." + - id: "candidate_worker" + display_name: "Builder-Vitruvius" + role: "Exercise actual workflow worker activation." model: model_gateway: "hasna" provider: "mock_provider" @@ -894,10 +1007,20 @@ agents: reasoning: "high" approval_policy: "never" permission_profile: "read-only" + - id: "adversarial_reviewer" + display_name: "Reviewer-Hypatia" + role: "Independently review the exact activation candidate." + model: + model_gateway: "hasna" + provider: "mock_provider" + model: "mock-model" + reasoning: "high" + approval_policy: "never" + permission_profile: "workspace-write" steps: - - id: "adversarial_actual_worker" - title: "Run the actual adversarial worker" - agent: "adversarial_worker" + - id: "actual_worker" + title: "Run the actual workflow worker" + agent: "candidate_worker" model: model_gateway: "hasna" provider: "mock_provider" @@ -922,9 +1045,40 @@ steps: commands: - "git status --short" expected_exit_code: 0 + - id: "initial_adversarial_review" + title: "Run the initial adversarial review" + agent: "adversarial_reviewer" + model: + model_gateway: "hasna" + provider: "mock_provider" + model: "mock-model" + reasoning: "high" + approval_policy: "never" + permission_profile: "workspace-write" + workspace: + mode: "isolated_worktree" + depends_on: + - "actual_worker" + outputs: + - "review.yaml" + completion: + model_marked_state: "candidate_succeeded" + verifiers: + - id: "finite_review_artifact_contract" + type: "artifact_contains" + artifact: "review.yaml" + must_contain: + - "candidate_identity:" + - "acceptance_criteria:" + - "verdict:" + - "blocking_p0_p1:" + - "non_blocking_p2_p3:" + - "remediation_cycle:" + - "remediation_cycle_cap: 2" artifacts: retention: "preserve_evidence" - required: [] + required: + - "review.yaml" cleanup: on_cancel: [] on_complete: [] diff --git a/codex-rs/ext/workflows/Cargo.toml b/codex-rs/ext/workflows/Cargo.toml index aa94f82314..d0ae2a8245 100644 --- a/codex-rs/ext/workflows/Cargo.toml +++ b/codex-rs/ext/workflows/Cargo.toml @@ -26,11 +26,21 @@ codex-utils-absolute-path = { workspace = true } codex-workflows = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +serde_yaml = { workspace = true } sha2 = { workspace = true } tokio = { workspace = true, features = ["rt", "sync", "time"] } tokio-util = { workspace = true, features = ["rt"] } tracing = { workspace = true } +[target.'cfg(unix)'.dependencies] +libc = { workspace = true } + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.52", features = [ + "Win32_Foundation", + "Win32_Storage_FileSystem", +] } + [dev-dependencies] chrono = { workspace = true } codex-prompts = { workspace = true } diff --git a/codex-rs/ext/workflows/src/activation.rs b/codex-rs/ext/workflows/src/activation.rs index 2a1c29b91d..f0095ad4d1 100644 --- a/codex-rs/ext/workflows/src/activation.rs +++ b/codex-rs/ext/workflows/src/activation.rs @@ -1,6 +1,9 @@ use std::collections::HashMap; use std::collections::HashSet; +use std::fs::File; use std::future::Future; +use std::io::Read; +use std::path::Component; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -32,6 +35,7 @@ use codex_state::WorkflowRunFenceParams; use codex_state::WorkflowRunHeartbeatParams; use codex_state::WorkflowRunSnapshot; use codex_state::WorkflowRunStatus; +use codex_state::WorkflowRunStepVerifierStatus; use codex_state::WorkflowRunVerifierClaimOutcome; use codex_state::WorkflowRunVerifierClaimParams; use codex_state::WorkflowRunVerifierClaimSelection; @@ -41,6 +45,8 @@ use codex_state::WorkflowRunVerifierResultSummary; use codex_state::busy_retry::retry_on_busy; use codex_utils_absolute_path::AbsolutePathBuf; use codex_workflows::WorkflowVerifier; +use codex_workflows::verifier_has_finite_review_artifact_contract; +use serde::Deserialize; use serde::Serialize; use sha2::Digest; use sha2::Sha256; @@ -49,6 +55,37 @@ use tokio_util::sync::CancellationToken; const WORKFLOW_SUPERVISOR_POLL_INTERVAL: Duration = Duration::from_millis(250); const WORKFLOW_HEARTBEAT_INTERVAL: Duration = Duration::from_millis(500); +const ARTIFACT_CONTAINS_VERIFIER: &str = "artifact_contains"; +const RUN_COMMANDS_VERIFIER: &str = "run_commands"; +const MAX_ARTIFACT_CONTAINS_BYTES: u64 = 1024 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ArtifactContainsEvaluation { + artifact: PathBuf, + artifact_sha256: String, + byte_len: u64, + passed: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct FiniteReviewRuntimeContext { + candidate_identity: String, + acceptance_criteria: Vec, + artifact: String, + artifact_preexisting: bool, + admitted_at_ms: i64, +} + +#[derive(Debug, Deserialize)] +struct FiniteReviewArtifact { + candidate_identity: String, + acceptance_criteria: Vec, + verdict: String, + blocking_p0_p1: Vec, + non_blocking_p2_p3: Vec, + remediation_cycle: u8, + remediation_cycle_cap: u8, +} #[cfg(target_os = "linux")] const VERIFIER_SANDBOX_SUPPORT_ENV_VARS: [&str; 5] = [ @@ -449,6 +486,30 @@ impl WorkflowActivationService { continue; } + if let Some(verifier_run_id) = next_artifact_contains_verifier(&admitted.snapshot) { + let verifier_claim_params = WorkflowRunVerifierClaimParams { + run_id: run_id.to_string(), + owner_id: self.owner_instance_id.to_string(), + generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId(verifier_run_id), + }; + if let Some(claimed_verifier) = + retry_workflow_state(WorkflowStateOperation::ClaimVerifier, || { + self.state_db + .claim_workflow_run_verifier(verifier_claim_params.clone()) + }) + .await? + { + if !self + .execute_verifier(run_id, generation, claimed_verifier, config) + .await? + { + return Ok(()); + } + continue; + } + } + if !(advanced.changed || reconciled.changed || admitted.changed) { tokio::time::sleep(WORKFLOW_SUPERVISOR_POLL_INTERVAL).await; } @@ -480,6 +541,30 @@ impl WorkflowActivationService { .await; } }; + if definition.kind == ARTIFACT_CONTAINS_VERIFIER { + return self + .execute_artifact_contains_verifier( + run_id, + generation, + claimed, + &definition, + started, + ) + .await; + } + if definition.kind != RUN_COMMANDS_VERIFIER { + tracing::warn!( + workflow_run_id = %run_id, + verifier_run_id = %claimed.verifier.verifier_run_id, + verifier_type = %definition.kind, + "workflow verifier type is unsupported by the activation service" + ); + return self + .record_failed_verifier_setup( + run_id, generation, claimed, started, /*expected_exit_code*/ None, + ) + .await; + } let expected_exit_code = definition.expected_exit_code.unwrap_or(0); let execution_root = match verifier_execution_root(&claimed) { Ok(execution_root) => execution_root, @@ -677,6 +762,87 @@ impl WorkflowActivationService { Ok(recorded.is_some()) } + async fn execute_artifact_contains_verifier( + &self, + run_id: &str, + generation: i64, + claimed: WorkflowRunVerifierClaimOutcome, + definition: &WorkflowVerifier, + started: std::time::Instant, + ) -> anyhow::Result { + let evaluation = + finite_review_runtime_context(&claimed, definition).and_then(|review_context| { + verifier_execution_root(&claimed).and_then(|execution_root| { + evaluate_artifact_contains(&execution_root, definition, review_context.as_ref()) + }) + }); + let (outcome, output_bytes) = match evaluation { + Ok(evaluation) => { + tracing::debug!( + workflow_run_id = %run_id, + verifier_run_id = %claimed.verifier.verifier_run_id, + artifact = %evaluation.artifact.display(), + artifact_sha256 = %evaluation.artifact_sha256, + matched = evaluation.passed, + "workflow artifact_contains verifier evaluated a bounded artifact" + ); + ( + if evaluation.passed { + WorkflowRunVerifierOutcomeStatus::Passed + } else { + WorkflowRunVerifierOutcomeStatus::Failed + }, + i64::try_from(evaluation.byte_len).unwrap_or(i64::MAX), + ) + } + Err(err) => { + tracing::warn!( + workflow_run_id = %run_id, + verifier_run_id = %claimed.verifier.verifier_run_id, + "workflow artifact_contains verifier failed: {err}" + ); + (WorkflowRunVerifierOutcomeStatus::Failed, 0) + } + }; + + let fence_params = WorkflowRunFenceParams { + run_id: run_id.to_string(), + owner_id: self.owner_instance_id.to_string(), + generation, + }; + if !retry_workflow_state(WorkflowStateOperation::CheckVerifierFence, || { + self.state_db + .workflow_run_fence_is_current(fence_params.clone()) + }) + .await? + { + return Ok(false); + } + + let record_params = WorkflowRunVerifierRecordResultParams { + run_id: run_id.to_string(), + owner_id: self.owner_instance_id.to_string(), + generation, + verifier_run_id: claimed.verifier.verifier_run_id, + outcome, + summary: WorkflowRunVerifierResultSummary { + command_count: 0, + expected_exit_code: None, + observed_exit_code: None, + timed_out: false, + duration_ms: i64::try_from(started.elapsed().as_millis()).unwrap_or(i64::MAX), + output_bytes, + output_truncated: false, + }, + }; + let recorded = retry_workflow_state(WorkflowStateOperation::RecordVerifierResult, || { + self.state_db + .record_workflow_run_verifier_result(record_params.clone()) + }) + .await?; + Ok(recorded.is_some()) + } + async fn record_failed_verifier_setup( &self, run_id: &str, @@ -747,6 +913,423 @@ impl WorkflowActivationService { } } +fn next_artifact_contains_verifier(snapshot: &WorkflowRunSnapshot) -> Option { + snapshot + .verifiers + .iter() + .find(|verifier| { + verifier.verifier_type == ARTIFACT_CONTAINS_VERIFIER + && verifier.status == WorkflowRunStepVerifierStatus::Blocked + }) + .map(|verifier| verifier.verifier_run_id.clone()) +} + +fn evaluate_artifact_contains( + execution_root: &AbsolutePathBuf, + definition: &WorkflowVerifier, + review_context: Option<&FiniteReviewRuntimeContext>, +) -> anyhow::Result { + let artifact = definition + .artifact + .as_deref() + .ok_or_else(|| anyhow::anyhow!("artifact_contains verifier is missing artifact"))?; + let artifact_path = Path::new(artifact); + let (artifact, mut file) = securely_open_workspace_artifact(execution_root, artifact_path)?; + let metadata = file.metadata()?; + if !metadata.is_file() { + anyhow::bail!("workflow artifact_contains artifact is not a regular file"); + } + if metadata.len() > MAX_ARTIFACT_CONTAINS_BYTES { + anyhow::bail!( + "workflow artifact_contains artifact exceeds {MAX_ARTIFACT_CONTAINS_BYTES} bytes" + ); + } + let mut bytes = Vec::with_capacity(usize::try_from(metadata.len()).unwrap_or(64 * 1024)); + (&mut file) + .take(MAX_ARTIFACT_CONTAINS_BYTES + 1) + .read_to_end(&mut bytes)?; + if u64::try_from(bytes.len()).unwrap_or(u64::MAX) > MAX_ARTIFACT_CONTAINS_BYTES { + anyhow::bail!( + "workflow artifact_contains artifact exceeds {MAX_ARTIFACT_CONTAINS_BYTES} bytes" + ); + } + let metadata_after_read = file.metadata()?; + let modified_while_read = match (metadata.modified(), metadata_after_read.modified()) { + (Ok(before), Ok(after)) => before != after, + _ => false, + }; + if metadata_after_read.len() != metadata.len() || modified_while_read { + anyhow::bail!("workflow artifact_contains artifact changed while it was read"); + } + let content = std::str::from_utf8(&bytes) + .map_err(|_| anyhow::anyhow!("workflow artifact_contains artifact must be UTF-8"))?; + let passed = if verifier_has_finite_review_artifact_contract(definition) { + evaluate_finite_review_artifact( + artifact_path, + &metadata, + content, + definition, + review_context, + )? + } else { + definition + .must_contain + .iter() + .all(|required| content.contains(required)) + }; + Ok(ArtifactContainsEvaluation { + artifact, + artifact_sha256: format!("{:x}", Sha256::digest(&bytes)), + byte_len: u64::try_from(bytes.len()).unwrap_or(u64::MAX), + passed, + }) +} + +fn normalized_artifact_components(artifact: &Path) -> anyhow::Result> { + if artifact.is_absolute() { + anyhow::bail!("workflow artifact_contains artifact must be relative to the workspace"); + } + let components = artifact + .components() + .map(|component| match component { + Component::Normal(value) => Ok(value), + _ => Err(anyhow::anyhow!( + "workflow artifact_contains artifact escapes the admitted workspace" + )), + }) + .collect::>>()?; + if components.is_empty() { + anyhow::bail!("workflow artifact_contains artifact has no file component"); + } + Ok(components) +} + +#[cfg(unix)] +fn securely_open_workspace_artifact( + execution_root: &AbsolutePathBuf, + artifact: &Path, +) -> anyhow::Result<(PathBuf, File)> { + use std::ffi::CString; + use std::os::fd::AsRawFd; + use std::os::fd::FromRawFd; + use std::os::unix::ffi::OsStrExt; + + let artifact_components = normalized_artifact_components(artifact)?; + let root = std::fs::canonicalize(execution_root.as_path())?; + if !root.is_absolute() { + anyhow::bail!("workflow artifact_contains execution root must be absolute"); + } + let root_components = root + .components() + .filter_map(|component| match component { + Component::RootDir => None, + Component::Normal(value) => Some(Ok(value)), + _ => Some(Err(anyhow::anyhow!( + "workflow artifact_contains execution root is not canonical" + ))), + }) + .collect::>>()?; + let mut components = root_components; + components.extend(artifact_components); + + // SAFETY: the constant path is NUL-terminated and the returned descriptor + // is immediately owned by `File` on success. + let root_fd = unsafe { + libc::open( + c"/".as_ptr(), + libc::O_RDONLY | libc::O_CLOEXEC | libc::O_DIRECTORY, + ) + }; + if root_fd < 0 { + return Err(std::io::Error::last_os_error().into()); + } + // SAFETY: `root_fd` is a fresh, owned descriptor from `open`. + let mut current = unsafe { File::from_raw_fd(root_fd) }; + for (index, component) in components.iter().enumerate() { + let component = CString::new(component.as_bytes()) + .map_err(|_| anyhow::anyhow!("workflow artifact path component contains NUL"))?; + let is_last = index + 1 == components.len(); + let flags = if is_last { + libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK + } else { + libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_DIRECTORY + }; + // SAFETY: `current` is a live directory descriptor for intermediate + // components and `component` is a NUL-terminated path component. + let next_fd = unsafe { libc::openat(current.as_raw_fd(), component.as_ptr(), flags) }; + if next_fd < 0 { + return Err(anyhow::anyhow!( + "cannot securely open workflow artifact {}: {}", + root.join(artifact).display(), + std::io::Error::last_os_error() + )); + } + // SAFETY: `next_fd` is a fresh, owned descriptor from `openat`. + current = unsafe { File::from_raw_fd(next_fd) }; + } + if !current.metadata()?.file_type().is_file() { + anyhow::bail!("workflow artifact_contains artifact is not a regular file"); + } + Ok((root.join(artifact), current)) +} + +#[cfg(windows)] +fn securely_open_workspace_artifact( + execution_root: &AbsolutePathBuf, + artifact: &Path, +) -> anyhow::Result<(PathBuf, File)> { + use std::os::windows::fs::MetadataExt; + use std::os::windows::fs::OpenOptionsExt; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Storage::FileSystem::FILE_ATTRIBUTE_REPARSE_POINT; + use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_BACKUP_SEMANTICS; + use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OPEN_REPARSE_POINT; + use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_DELETE; + use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ; + use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_WRITE; + + let artifact_components = normalized_artifact_components(artifact)?; + let root = std::fs::canonicalize(execution_root.as_path())?; + let root_file = std::fs::OpenOptions::new() + .access_mode(0) + .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE) + .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT) + .open(&root)?; + let opened_root = windows_final_path(root_file.as_raw_handle())?; + let artifact_path = root.join(artifact); + let mut component_path = root.clone(); + for (index, component) in artifact_components.iter().enumerate() { + component_path.push(component); + let metadata = std::fs::symlink_metadata(&component_path)?; + if metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 { + anyhow::bail!("workflow artifact_contains artifact contains a reparse point"); + } + if index + 1 != artifact_components.len() && !metadata.is_dir() { + anyhow::bail!("workflow artifact_contains artifact ancestor is not a directory"); + } + } + let file = std::fs::OpenOptions::new() + .read(true) + .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE) + .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT) + .open(&artifact_path)?; + let metadata = file.metadata()?; + if !metadata.file_type().is_file() { + anyhow::bail!("workflow artifact_contains artifact is not a regular file"); + } + let opened_artifact = windows_final_path(file.as_raw_handle())?; + if !opened_artifact.starts_with(&opened_root) { + anyhow::bail!("workflow artifact_contains artifact escapes the admitted workspace"); + } + Ok((artifact_path, file)) +} + +#[cfg(windows)] +fn windows_final_path(handle: std::os::windows::io::RawHandle) -> anyhow::Result { + use std::ffi::OsString; + use std::os::windows::ffi::OsStringExt; + use windows_sys::Win32::Foundation::HANDLE; + use windows_sys::Win32::Storage::FileSystem::FILE_NAME_NORMALIZED; + use windows_sys::Win32::Storage::FileSystem::GetFinalPathNameByHandleW; + + // SAFETY: `handle` is borrowed from a live `File`. A null buffer with a + // zero length requests the required UTF-16 buffer size. + let required = unsafe { + GetFinalPathNameByHandleW( + handle as HANDLE, + std::ptr::null_mut(), + 0, + FILE_NAME_NORMALIZED, + ) + }; + if required == 0 { + return Err(std::io::Error::last_os_error().into()); + } + let mut buffer = vec![0_u16; usize::try_from(required)?]; + // SAFETY: `buffer` is writable for its declared length and `handle` stays + // live for the duration of the call. + let written = unsafe { + GetFinalPathNameByHandleW( + handle as HANDLE, + buffer.as_mut_ptr(), + u32::try_from(buffer.len())?, + FILE_NAME_NORMALIZED, + ) + }; + if written == 0 { + return Err(std::io::Error::last_os_error().into()); + } + if usize::try_from(written)? >= buffer.len() { + anyhow::bail!("workflow artifact_contains final path changed while it was read"); + } + buffer.truncate(usize::try_from(written)?); + Ok(PathBuf::from(OsString::from_wide(&buffer))) +} + +#[cfg(not(any(unix, windows)))] +fn securely_open_workspace_artifact( + _execution_root: &AbsolutePathBuf, + _artifact: &Path, +) -> anyhow::Result<(PathBuf, File)> { + anyhow::bail!("workflow artifact_contains requires platform no-follow file semantics") +} + +fn finite_review_runtime_context( + claimed: &WorkflowRunVerifierClaimOutcome, + definition: &WorkflowVerifier, +) -> anyhow::Result> { + if !verifier_has_finite_review_artifact_contract(definition) { + return Ok(None); + } + let admission = claimed + .step + .branch_admission_json + .as_ref() + .map(workflow_state_data); + let review_context = admission.and_then(|value| value.get("reviewContext")); + let Some(review_context) = review_context else { + if claimed.step.background_agent_run_id.is_some() { + anyhow::bail!( + "finite review verifier {} has no persisted review context", + claimed.verifier.verifier_run_id + ); + } + return Ok(None); + }; + let candidate_identity = required_json_string(review_context, "candidateIdentity")?; + let acceptance_criteria = review_context + .get("acceptanceCriteria") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| anyhow::anyhow!("finite review context has no acceptanceCriteria"))? + .iter() + .map(|criterion| { + criterion + .as_str() + .map(str::trim) + .filter(|criterion| !criterion.is_empty()) + .map(str::to_string) + .ok_or_else(|| { + anyhow::anyhow!("finite review context has an invalid acceptance criterion") + }) + }) + .collect::>>()?; + if acceptance_criteria.is_empty() { + anyhow::bail!("finite review context has no acceptance criteria"); + } + let artifact = required_json_string(review_context, "artifact")?; + let definition_artifact = definition + .artifact + .as_deref() + .ok_or_else(|| anyhow::anyhow!("finite review verifier has no artifact"))?; + if artifact != definition_artifact { + anyhow::bail!( + "finite review context artifact `{artifact}` does not match verifier artifact `{definition_artifact}`" + ); + } + let artifact_preexisting = review_context + .get("artifactPreexisting") + .and_then(serde_json::Value::as_bool) + .ok_or_else(|| anyhow::anyhow!("finite review context has no artifactPreexisting flag"))?; + let admitted_at_ms = admission + .and_then(|value| value.get("admittedAtMs")) + .and_then(serde_json::Value::as_i64) + .ok_or_else(|| anyhow::anyhow!("finite review admission has no admittedAtMs"))?; + Ok(Some(FiniteReviewRuntimeContext { + candidate_identity, + acceptance_criteria, + artifact, + artifact_preexisting, + admitted_at_ms, + })) +} + +fn required_json_string(value: &serde_json::Value, field: &str) -> anyhow::Result { + let value = value + .get(field) + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .ok_or_else(|| anyhow::anyhow!("finite review context has no non-empty `{field}`"))?; + Ok(value.to_string()) +} + +fn evaluate_finite_review_artifact( + artifact_path: &Path, + metadata: &std::fs::Metadata, + content: &str, + definition: &WorkflowVerifier, + review_context: Option<&FiniteReviewRuntimeContext>, +) -> anyhow::Result { + let Ok(artifact_value) = serde_yaml::from_str::(content) else { + return Ok(false); + }; + let Ok(artifact) = serde_yaml::from_value::(artifact_value.clone()) + else { + return Ok(false); + }; + let Some(mapping) = artifact_value.as_mapping() else { + return Ok(false); + }; + if !definition + .must_contain + .iter() + .all(|required| yaml_mapping_satisfies_requirement(mapping, required)) + { + return Ok(false); + } + if artifact.candidate_identity.trim().is_empty() + || artifact.acceptance_criteria.is_empty() + || artifact + .acceptance_criteria + .iter() + .any(|criterion| criterion.trim().is_empty()) + || artifact.verdict != "GO" + || !artifact.blocking_p0_p1.is_empty() + || artifact.remediation_cycle > 2 + || artifact.remediation_cycle_cap != 2 + { + return Ok(false); + } + let _non_blocking_findings_are_typed = &artifact.non_blocking_p2_p3; + let Some(review_context) = review_context else { + return Ok(true); + }; + if review_context.artifact_preexisting + || review_context.artifact != artifact_path.to_string_lossy().as_ref() + || artifact.candidate_identity != review_context.candidate_identity + || artifact.acceptance_criteria != review_context.acceptance_criteria + { + return Ok(false); + } + let modified_at_ms = metadata + .modified()? + .duration_since(std::time::UNIX_EPOCH) + .map_err(|_| anyhow::anyhow!("finite review artifact modification time predates epoch"))? + .as_millis(); + let modified_at_ms = i64::try_from(modified_at_ms).unwrap_or(i64::MAX); + Ok(modified_at_ms >= review_context.admitted_at_ms) +} + +fn yaml_mapping_satisfies_requirement(mapping: &serde_yaml::Mapping, requirement: &str) -> bool { + let requirement = requirement.trim(); + let Some((field, expected)) = requirement.split_once(':') else { + return false; + }; + let field = field.trim(); + if field.is_empty() { + return false; + } + let key = serde_yaml::Value::String(field.to_string()); + let Some(actual) = mapping.get(&key) else { + return false; + }; + let expected = expected.trim(); + if expected.is_empty() { + return true; + } + serde_yaml::from_str::(expected).is_ok_and(|expected| expected == *actual) +} + #[derive(Serialize)] #[serde(rename_all = "camelCase")] struct WorkflowActivationFingerprint<'a> { @@ -962,3 +1545,7 @@ mod tests { assert!(!env.contains_key("UNRELATED_ENV")); } } + +#[cfg(test)] +#[path = "activation/finance_acceptance.rs"] +mod finance_acceptance; diff --git a/codex-rs/ext/workflows/src/activation/finance_acceptance.rs b/codex-rs/ext/workflows/src/activation/finance_acceptance.rs new file mode 100644 index 0000000000..416c6117df --- /dev/null +++ b/codex-rs/ext/workflows/src/activation/finance_acceptance.rs @@ -0,0 +1,1115 @@ +use super::*; + +use chrono::Utc; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionSource; +use codex_state::ThreadGoalPlanAutoExecute; +use codex_state::ThreadGoalPlanNodeCompletionStatus; +use codex_state::ThreadGoalPlanNodeStatusUpdateParams; +use codex_state::ThreadMetadataBuilder; +use codex_state::WorkflowRunStepStatus; +use codex_state::WorkflowSpecCreateParams; +use codex_workflows::WorkflowSpec; +use serde::Serialize; +use serde_json::Value; +use std::path::Path; +use std::path::PathBuf; +use std::process::Command; +use std::sync::Arc; +use tempfile::TempDir; + +const CANDIDATE_STEP_ID: &str = "build_candidate"; +const REVIEW_STEP_ID: &str = "initial_adversarial_review"; +const REVIEW_VERIFIER_ID: &str = "finite_review_artifact_contract"; +const REVIEW_ARTIFACT: &str = "review.yaml"; +const FINITE_REVIEW_FIELDS: [&str; 9] = [ + "candidate_identity:", + "acceptance_criteria:", + "verdict:", + "blocking_p0_p1:", + "non_blocking_p2_p3:", + "remediation_cycle:", + "remediation_cycle_cap: 2", + "payment_provider_execution_steps: 0", + "finance_provider_execution_calls: 0", +]; + +fn synthetic_finance_workflow_yaml(workflow_id: &str) -> String { + format!( + r#"schema_version: "workflow.codex.codewith/v0" +workflow_id: "{workflow_id}" +display_name: "Synthetic finance acceptance" +source_prompt: "Build a deterministic analysis artifact and run one adversarial review without finance execution." +status: "draft" +execution_defaults: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" +limits: + max_parallel_steps: 1 + max_agents: 2 + max_worktrees: 1 + max_runtime_seconds: 3600 + max_step_runtime_seconds: 1200 + max_tokens: 100000 + max_tool_calls: 100 +approvals: + required_before: [] +agents: + - id: "candidate_builder" + display_name: "Builder-Vitruvius" + role: "Build the deterministic analysis artifact and record its acceptance criteria." + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" + - id: "adversarial_reviewer" + display_name: "Reviewer-Hypatia" + role: "Independently and adversarially review the exact analysis artifact." + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" +steps: + - id: "{CANDIDATE_STEP_ID}" + title: "Build the deterministic analysis artifact" + agent: "candidate_builder" + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" + depends_on: [] + outputs: + - "candidate.txt" + completion: + model_marked_state: "candidate_succeeded" + verifiers: + - id: "candidate_identity_present" + type: "artifact_contains" + artifact: "candidate.txt" + must_contain: + - "candidate_identity:" + - id: "{REVIEW_STEP_ID}" + title: "Run the initial adversarial review" + agent: "adversarial_reviewer" + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" + depends_on: + - "{CANDIDATE_STEP_ID}" + outputs: + - "{REVIEW_ARTIFACT}" + completion: + model_marked_state: "candidate_succeeded" + verifiers: + - id: "{REVIEW_VERIFIER_ID}" + type: "artifact_contains" + artifact: "{REVIEW_ARTIFACT}" + must_contain: + - "candidate_identity:" + - "acceptance_criteria:" + - "verdict:" + - "blocking_p0_p1:" + - "non_blocking_p2_p3:" + - "remediation_cycle:" + - "remediation_cycle_cap: 2" + - "payment_provider_execution_steps: 0" + - "finance_provider_execution_calls: 0" +artifacts: + retention: "preserve_evidence" + required: + - "candidate.txt" + - "{REVIEW_ARTIFACT}" +cleanup: + on_cancel: [] + on_complete: [] +"#, + ) +} + +#[derive(Debug, Clone, Default, Serialize)] +struct FinanceBoundaryCounters { + payment_calls: u64, + approval_calls: u64, + scheduling_calls: u64, + transfer_calls: u64, + bank_provider_submission_calls: u64, + invoice_mutation_calls: u64, + finance_credential_calls: u64, + production_finance_writes: u64, +} + +impl FinanceBoundaryCounters { + fn total(&self) -> u64 { + self.payment_calls + + self.approval_calls + + self.scheduling_calls + + self.transfer_calls + + self.bank_provider_submission_calls + + self.invoice_mutation_calls + + self.finance_credential_calls + + self.production_finance_writes + } + + fn inject_bank_provider_submission_attempt(&mut self) -> anyhow::Result<()> { + self.bank_provider_submission_calls += 1; + anyhow::bail!("synthetic finance boundary rejected bank/provider submission before I/O") + } +} + +#[derive(Debug, Serialize)] +struct StepReceipt { + step_id: String, + completion_model_marked_state: String, + waiting_verifier_event_seq: i64, + verifier_passed_event_seq: i64, + succeeded_event_seq: i64, + succeeded_at: String, +} + +#[derive(Debug, Serialize)] +struct ArtifactContainsReceipt { + verifier_id: String, + verifier_type: String, + status: String, + required_fields: Vec, + artifact_id: String, + artifact_sha256: String, + artifact_bytes: u64, +} + +#[derive(Debug, Serialize)] +struct FinanceAcceptanceReceipt { + candidate_sha: String, + workflow_id: String, + workflow_source_sha256: String, + run_id: String, + run_started_at: String, + run_succeeded_at: String, + builder_count: usize, + reviewer_count: usize, + review_step_count: usize, + steps: Vec, + remediation_cycle: u8, + remediation_cycle_cap: u8, + payment_provider_execution_steps: usize, + finance_boundary_counters: FinanceBoundaryCounters, + model_provider_calls_excluded_from_finance_counters: bool, + artifact_contains: ArtifactContainsReceipt, + persistence_scope: &'static str, + external_persistence_writes: u64, +} + +struct SyntheticFinanceHarness { + _temp: TempDir, + runtime: Arc, + service: WorkflowActivationService, + thread_id: ThreadId, + workspace: PathBuf, + spec: WorkflowSpec, + source_yaml_sha256: String, + projection: WorkflowGoalPlanProjectionOutcome, + run_id: String, + generation: i64, +} + +struct PreparedVerifier { + verifier_run_id: String, + evaluation: ArtifactContainsEvaluation, + snapshot: WorkflowRunSnapshot, +} + +impl SyntheticFinanceHarness { + async fn new(workflow_id: &str) -> anyhow::Result { + let temp = tempfile::tempdir()?; + let codex_home = temp.path().join("codex-home"); + let workspace = temp.path().join("workspace"); + initialize_test_git_repo(&workspace)?; + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()).await?; + let thread_id = ThreadId::new(); + let mut metadata = ThreadMetadataBuilder::new( + thread_id, + codex_home.join(format!("rollout-{thread_id}.jsonl")), + Utc::now(), + SessionSource::Cli, + ); + metadata.cwd = workspace.clone(); + runtime + .upsert_thread(&metadata.build("test-provider")) + .await?; + + let yaml = synthetic_finance_workflow_yaml(workflow_id); + let spec = codex_workflows::parse_workflow_yaml(&yaml)?; + validate_synthetic_finance_structure(&spec)?; + let saved = runtime + .workflows() + .save_workflow_spec_yaml(WorkflowSpecCreateParams { + source_thread_id: Some(thread_id), + source_yaml: yaml, + }) + .await?; + let run = runtime + .workflows() + .create_workflow_run(WorkflowRunCreateParams { + workflow_record_id: saved.workflow_record_id, + source_thread_id: Some(thread_id), + idempotency_key: Some(format!("{workflow_id}-run")), + }) + .await?; + let projection = runtime + .project_workflow_run_to_goal_plan(WorkflowGoalPlanProjectionParams { + workflow_run_id: run.run.run_id.clone(), + thread_id, + idempotency_key: Some(format!("{workflow_id}-projection")), + }) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow run did not project to a goal plan"))?; + let service = WorkflowActivationService::new(Arc::clone(&runtime)); + let claim = runtime + .claim_workflow_run(WorkflowRunClaimParams { + run_id: run.run.run_id.clone(), + owner_id: service.owner_instance_id.to_string(), + lease_duration_ms: Some(60_000), + }) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow run was not claimable"))?; + + Ok(Self { + _temp: temp, + runtime, + service, + thread_id, + workspace, + spec, + source_yaml_sha256: run.run.source_yaml_sha256, + projection, + run_id: run.run.run_id, + generation: claim.generation, + }) + } + + async fn prepare_step( + &self, + step_id: &str, + artifact: &str, + content: &str, + ) -> anyhow::Result { + std::fs::write(self.workspace.join(artifact), content)?; + let node = self + .projection + .snapshot + .nodes + .iter() + .find(|node| node.key == step_id) + .ok_or_else(|| anyhow::anyhow!("projected node `{step_id}` is missing"))?; + self.runtime + .thread_goals() + .set_thread_goal_plan_node_status(ThreadGoalPlanNodeStatusUpdateParams { + thread_id: self.thread_id, + node_id: node.node_id.clone(), + status: ThreadGoalPlanNodeCompletionStatus::Complete, + auto_execute: ThreadGoalPlanAutoExecute::Off, + }) + .await?; + let advanced = self + .runtime + .advance_workflow_run(WorkflowRunAdvanceParams { + run_id: self.run_id.clone(), + owner_id: self.service.owner_instance_id.to_string(), + generation: self.generation, + }) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow run could not advance"))?; + let step = advanced + .snapshot + .steps + .iter() + .find(|step| step.step_id == step_id) + .ok_or_else(|| anyhow::anyhow!("workflow step `{step_id}` is missing"))?; + anyhow::ensure!( + step.completion_model_marked_state.as_deref() == Some("candidate_succeeded"), + "workflow step `{step_id}` lost candidate_succeeded completion state" + ); + anyhow::ensure!( + step.status == WorkflowRunStepStatus::WaitingVerifier, + "workflow step `{step_id}` did not reach waiting_verifier" + ); + let verifier = advanced + .snapshot + .verifiers + .iter() + .find(|verifier| verifier.step_id == step_id) + .ok_or_else(|| anyhow::anyhow!("workflow step `{step_id}` has no verifier"))?; + anyhow::ensure!( + verifier.status == WorkflowRunStepVerifierStatus::Blocked, + "workflow verifier for `{step_id}` was not verifier-gated" + ); + let definition: WorkflowVerifier = + serde_json::from_value(workflow_state_data(&verifier.definition_json).clone())?; + let execution_root = AbsolutePathBuf::try_from(self.workspace.canonicalize()?) + .map_err(|err| anyhow::anyhow!("test workspace is invalid: {err}"))?; + let evaluation = + evaluate_artifact_contains(&execution_root, &definition, /*review_context*/ None)?; + Ok(PreparedVerifier { + verifier_run_id: verifier.verifier_run_id.clone(), + evaluation, + snapshot: advanced.snapshot, + }) + } + + async fn execute_prepared( + &self, + prepared: PreparedVerifier, + ) -> anyhow::Result { + let claimed = self + .runtime + .claim_workflow_run_verifier(WorkflowRunVerifierClaimParams { + run_id: self.run_id.clone(), + owner_id: self.service.owner_instance_id.to_string(), + generation: self.generation, + selection: WorkflowRunVerifierClaimSelection::VerifierRunId( + prepared.verifier_run_id, + ), + }) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow verifier was not claimable"))?; + anyhow::ensure!( + self.service + .execute_verifier( + &self.run_id, + self.generation, + claimed, + &WorkflowActivationConfig::default(), + ) + .await?, + "workflow verifier result was not persisted" + ); + self.runtime + .workflows() + .get_workflow_run_snapshot(&self.run_id) + .await? + .ok_or_else(|| anyhow::anyhow!("workflow run snapshot disappeared")) + } + + async fn complete_candidate(&self, candidate_sha: &str) -> anyhow::Result<()> { + let prepared = self + .prepare_step( + CANDIDATE_STEP_ID, + "candidate.txt", + &format!("candidate_identity: {candidate_sha}\n"), + ) + .await?; + anyhow::ensure!( + prepared.evaluation.passed, + "candidate artifact did not match" + ); + let snapshot = self.execute_prepared(prepared).await?; + let step = snapshot + .steps + .iter() + .find(|step| step.step_id == CANDIDATE_STEP_ID) + .ok_or_else(|| anyhow::anyhow!("candidate step disappeared"))?; + anyhow::ensure!( + step.status == WorkflowRunStepStatus::Succeeded, + "candidate step did not succeed" + ); + Ok(()) + } +} + +fn initialize_test_git_repo(path: &Path) -> anyhow::Result<()> { + std::fs::create_dir_all(path)?; + run_test_git(path, &["init"])?; + std::fs::write(path.join("README.md"), "synthetic finance acceptance\n")?; + run_test_git(path, &["add", "README.md"])?; + run_test_git( + path, + &[ + "commit", + "--no-gpg-sign", + "--no-verify", + "-m", + "Initialize synthetic finance fixture", + ], + )?; + Ok(()) +} + +fn run_test_git(path: &Path, args: &[&str]) -> anyhow::Result<()> { + let output = Command::new("git") + .args(args) + .current_dir(path) + .env("GIT_AUTHOR_NAME", "Codewith Test") + .env("GIT_AUTHOR_EMAIL", "codewith-test@example.invalid") + .env("GIT_COMMITTER_NAME", "Codewith Test") + .env("GIT_COMMITTER_EMAIL", "codewith-test@example.invalid") + .output()?; + anyhow::ensure!( + output.status.success(), + "git {:?} failed: {}", + args, + String::from_utf8_lossy(&output.stderr) + ); + Ok(()) +} + +fn validate_synthetic_finance_structure(spec: &WorkflowSpec) -> anyhow::Result<()> { + let reviewer_count = spec + .agents + .iter() + .filter(|agent| { + let text = + format!("{} {} {}", agent.id, agent.display_name, agent.role).to_ascii_lowercase(); + text.contains("review") || text.contains("adversarial") + }) + .count(); + let review_step_count = spec + .steps + .iter() + .filter(|step| { + let text = format!("{} {}", step.id, step.title).to_ascii_lowercase(); + text.contains("review") || text.contains("adversarial") + }) + .count(); + let payment_provider_execution_steps = payment_provider_execution_step_count(spec); + anyhow::ensure!( + spec.agents.len() == 2, + "acceptance workflow must have two agents" + ); + anyhow::ensure!( + spec.steps.len() == 2, + "acceptance workflow must have two steps" + ); + anyhow::ensure!( + reviewer_count == 1, + "acceptance workflow must have one reviewer" + ); + anyhow::ensure!( + review_step_count == 1, + "acceptance workflow must have one review step" + ); + anyhow::ensure!( + payment_provider_execution_steps == 0, + "acceptance workflow contains {payment_provider_execution_steps} payment/provider execution step(s)" + ); + anyhow::ensure!( + spec.steps.iter().all(|step| step.approval_gate.is_none()), + "acceptance workflow cannot contain an approval gate" + ); + anyhow::ensure!( + spec.steps.iter().all(|step| { + step.completion.as_ref().is_some_and(|completion| { + !completion.verifiers.is_empty() + && completion + .verifiers + .iter() + .all(|verifier| verifier.kind == ARTIFACT_CONTAINS_VERIFIER) + }) + }), + "acceptance workflow permits artifact_contains verifiers only" + ); + Ok(()) +} + +fn payment_provider_execution_step_count(spec: &WorkflowSpec) -> usize { + const FORBIDDEN_TERMS: [&str; 12] = [ + "pay", + "payment", + "approve", + "approval", + "schedule", + "transfer", + "submit", + "bank", + "provider", + "invoice", + "credential", + "production finance", + ]; + spec.steps + .iter() + .filter(|step| { + let text = format!("{} {} {}", step.id, step.title, step.outputs.join(" ")) + .to_ascii_lowercase(); + let tokens = text + .split(|ch: char| !ch.is_ascii_alphanumeric()) + .filter(|token| !token.is_empty()) + .collect::>(); + FORBIDDEN_TERMS.iter().any(|term| { + if term.contains(' ') { + text.contains(term) + } else { + tokens.iter().any(|token| token == term) + } + }) + }) + .count() +} + +fn event_sequence( + snapshot: &WorkflowRunSnapshot, + event_type: &str, + step_id: &str, +) -> anyhow::Result { + snapshot + .events + .iter() + .find(|event| { + event.event_type == event_type + && workflow_state_data(&event.event_payload_json) + .get("stepId") + .and_then(Value::as_str) + == Some(step_id) + }) + .map(|event| event.seq) + .ok_or_else(|| anyhow::anyhow!("event `{event_type}` for `{step_id}` is missing")) +} + +fn terminal_acceptance_receipt( + harness: &SyntheticFinanceHarness, + snapshot: &WorkflowRunSnapshot, + candidate_sha: &str, + counters: FinanceBoundaryCounters, + review_evaluation: &ArtifactContainsEvaluation, +) -> anyhow::Result { + anyhow::ensure!( + snapshot.run.status == WorkflowRunStatus::Completed, + "terminal verifier-gated workflow receipt is missing" + ); + anyhow::ensure!( + counters.total() == 0, + "finance boundary counters must remain zero" + ); + let reviewer_count = harness + .spec + .agents + .iter() + .filter(|agent| agent.id == "adversarial_reviewer") + .count(); + let review_step_count = harness + .spec + .steps + .iter() + .filter(|step| step.id == REVIEW_STEP_ID) + .count(); + anyhow::ensure!(reviewer_count == 1, "receipt requires exactly one reviewer"); + anyhow::ensure!( + review_step_count == 1, + "receipt requires exactly one review step" + ); + let mut steps = Vec::new(); + for step_id in [CANDIDATE_STEP_ID, REVIEW_STEP_ID] { + let step = snapshot + .steps + .iter() + .find(|step| step.step_id == step_id) + .ok_or_else(|| anyhow::anyhow!("receipt step `{step_id}` is missing"))?; + anyhow::ensure!( + step.status == WorkflowRunStepStatus::Succeeded, + "receipt step `{step_id}` did not succeed" + ); + let waiting = event_sequence(snapshot, "step_waiting_verifier", step_id)?; + let passed = event_sequence(snapshot, "verifier_passed", step_id)?; + let succeeded = event_sequence(snapshot, "step_succeeded", step_id)?; + anyhow::ensure!( + waiting < passed && passed < succeeded, + "receipt step `{step_id}` did not follow candidate_succeeded -> verifier pass -> succeeded" + ); + steps.push(StepReceipt { + step_id: step_id.to_string(), + completion_model_marked_state: step + .completion_model_marked_state + .clone() + .ok_or_else(|| anyhow::anyhow!("step completion state is missing"))?, + waiting_verifier_event_seq: waiting, + verifier_passed_event_seq: passed, + succeeded_event_seq: succeeded, + succeeded_at: step + .completed_at + .ok_or_else(|| anyhow::anyhow!("step success timestamp is missing"))? + .to_rfc3339(), + }); + } + let review_verifier = snapshot + .verifiers + .iter() + .find(|verifier| verifier.verifier_id == REVIEW_VERIFIER_ID) + .ok_or_else(|| anyhow::anyhow!("review verifier is missing"))?; + anyhow::ensure!( + review_verifier.status == WorkflowRunStepVerifierStatus::Passed, + "review verifier did not pass" + ); + let result = review_verifier + .last_result_json + .as_ref() + .map(workflow_state_data) + .ok_or_else(|| anyhow::anyhow!("artifact_contains result receipt is missing"))?; + anyhow::ensure!( + result.get("status").and_then(Value::as_str) == Some("passed"), + "artifact_contains result receipt did not record passed" + ); + anyhow::ensure!(review_evaluation.passed, "review artifact did not match"); + let payment_provider_execution_steps = payment_provider_execution_step_count(&harness.spec); + anyhow::ensure!( + payment_provider_execution_steps == 0, + "receipt observed a payment/provider execution step" + ); + Ok(FinanceAcceptanceReceipt { + candidate_sha: candidate_sha.to_string(), + workflow_id: snapshot.run.spec_workflow_id.clone(), + workflow_source_sha256: harness.source_yaml_sha256.clone(), + run_id: snapshot.run.run_id.clone(), + run_started_at: snapshot + .run + .started_at + .ok_or_else(|| anyhow::anyhow!("run start timestamp is missing"))? + .to_rfc3339(), + run_succeeded_at: snapshot + .run + .completed_at + .ok_or_else(|| anyhow::anyhow!("run completion timestamp is missing"))? + .to_rfc3339(), + builder_count: harness + .spec + .agents + .iter() + .filter(|agent| agent.id == "candidate_builder") + .count(), + reviewer_count, + review_step_count, + steps, + remediation_cycle: 0, + remediation_cycle_cap: 2, + payment_provider_execution_steps, + finance_boundary_counters: counters, + model_provider_calls_excluded_from_finance_counters: true, + artifact_contains: ArtifactContainsReceipt { + verifier_id: review_verifier.verifier_id.clone(), + verifier_type: review_verifier.verifier_type.clone(), + status: review_verifier.status.as_str().to_string(), + required_fields: FINITE_REVIEW_FIELDS + .iter() + .map(|field| (*field).to_string()) + .collect(), + artifact_id: REVIEW_ARTIFACT.to_string(), + artifact_sha256: review_evaluation.artifact_sha256.clone(), + artifact_bytes: review_evaluation.byte_len, + }, + persistence_scope: "isolated temporary workflow runtime, goal-plan projection, run/step/verifier events, and deterministic artifacts", + external_persistence_writes: 0, + }) +} + +fn review_artifact(candidate_sha: &str, verdict: Option<&str>, finance_calls: u64) -> String { + let verdict = verdict + .map(|value| format!("verdict: {value}\n")) + .unwrap_or_default(); + format!( + "candidate_identity: {candidate_sha}\nacceptance_criteria:\n - exact workflow validator\n - persisted state transitions\n - artifact verifier\n{verdict}blocking_p0_p1: []\nnon_blocking_p2_p3: []\nremediation_cycle: 0\nremediation_cycle_cap: 2\npayment_provider_execution_steps: 0\nfinance_provider_execution_calls: {finance_calls}\n" + ) +} + +fn artifact_verifier(artifact: &str, must_contain: &[&str]) -> WorkflowVerifier { + WorkflowVerifier { + id: "artifact_control".to_string(), + kind: ARTIFACT_CONTAINS_VERIFIER.to_string(), + artifact: Some(artifact.to_string()), + must_contain: must_contain + .iter() + .map(|value| (*value).to_string()) + .collect(), + cwd: None, + sandbox: None, + network: None, + timeout_seconds: None, + output_limit_bytes: None, + commands: Vec::new(), + expected_stdout: None, + expected_exit_code: None, + retry_policy: None, + } +} + +#[test] +fn artifact_contains_is_bounded_to_the_admitted_workspace() { + let temp = tempfile::tempdir().expect("artifact control tempdir should create"); + let workspace = temp.path().join("workspace"); + std::fs::create_dir(&workspace).expect("artifact control workspace should create"); + std::fs::write(temp.path().join("outside.txt"), "required\n") + .expect("outside control should write"); + let execution_root = + AbsolutePathBuf::try_from(workspace.canonicalize().expect("canonical root")) + .expect("absolute root"); + + let escape = evaluate_artifact_contains( + &execution_root, + &artifact_verifier("../outside.txt", &["required"]), + /*review_context*/ None, + ) + .expect_err("artifact traversal must be rejected"); + assert!(escape.to_string().contains("escapes")); + + let missing = evaluate_artifact_contains( + &execution_root, + &artifact_verifier("missing.txt", &["required"]), + /*review_context*/ None, + ); + assert!(missing.is_err(), "missing artifacts must fail closed"); + + std::fs::write( + workspace.join("oversized.txt"), + vec![b'x'; usize::try_from(MAX_ARTIFACT_CONTAINS_BYTES + 1).expect("bounded size")], + ) + .expect("oversized control should write"); + let oversized = evaluate_artifact_contains( + &execution_root, + &artifact_verifier("oversized.txt", &["required"]), + /*review_context*/ None, + ) + .expect_err("oversized artifacts must be rejected"); + assert!(oversized.to_string().contains("exceeds")); +} + +#[cfg(unix)] +#[test] +fn artifact_contains_rejects_symlink_components() { + use std::os::unix::fs::symlink; + + let temp = tempfile::tempdir().expect("artifact symlink tempdir should create"); + let workspace = temp.path().join("workspace"); + let real = workspace.join("real"); + std::fs::create_dir(&workspace).expect("artifact symlink workspace should create"); + std::fs::create_dir(&real).expect("artifact real directory should create"); + std::fs::write(real.join("artifact.txt"), "required\n").expect("real artifact should write"); + let execution_root = + AbsolutePathBuf::try_from(workspace.canonicalize().expect("canonical root")) + .expect("absolute root"); + + symlink(real.join("artifact.txt"), workspace.join("final-link.txt")) + .expect("final symlink should create"); + assert!( + evaluate_artifact_contains( + &execution_root, + &artifact_verifier("final-link.txt", &["required"]), + /*review_context*/ None, + ) + .is_err(), + "a final symlink must fail closed" + ); + + symlink(&real, workspace.join("ancestor-link")).expect("ancestor symlink should create"); + assert!( + evaluate_artifact_contains( + &execution_root, + &artifact_verifier("ancestor-link/artifact.txt", &["required"]), + /*review_context*/ None, + ) + .is_err(), + "an ancestor symlink must fail closed" + ); +} + +#[cfg(unix)] +#[test] +fn artifact_contains_reads_the_opened_handle_after_path_swap() { + let temp = tempfile::tempdir().expect("artifact swap tempdir should create"); + let workspace = temp.path().join("workspace"); + std::fs::create_dir(&workspace).expect("artifact swap workspace should create"); + let artifact = workspace.join("artifact.txt"); + std::fs::write(&artifact, "original\n").expect("original artifact should write"); + let execution_root = + AbsolutePathBuf::try_from(workspace.canonicalize().expect("canonical root")) + .expect("absolute root"); + + let (_, mut opened) = + securely_open_workspace_artifact(&execution_root, Path::new("artifact.txt")) + .expect("artifact handle should open securely"); + std::fs::rename(&artifact, workspace.join("artifact-original.txt")) + .expect("opened artifact should rename"); + std::fs::write(&artifact, "replacement\n").expect("replacement artifact should write"); + + let mut content = String::new(); + std::io::Read::read_to_string(&mut opened, &mut content) + .expect("opened artifact should remain readable"); + assert_eq!(content, "original\n"); +} + +#[test] +fn finite_review_artifact_rejects_preexisting_and_stale_identity() { + let temp = tempfile::tempdir().expect("finite review tempdir should create"); + let workspace = temp.path().join("workspace"); + std::fs::create_dir(&workspace).expect("finite review workspace should create"); + std::fs::write( + workspace.join(REVIEW_ARTIFACT), + review_artifact("candidate-under-test", Some("GO"), /*finance_calls*/ 0), + ) + .expect("finite review artifact should write"); + let execution_root = + AbsolutePathBuf::try_from(workspace.canonicalize().expect("canonical root")) + .expect("absolute root"); + let definition = artifact_verifier(REVIEW_ARTIFACT, &FINITE_REVIEW_FIELDS); + + let preexisting = evaluate_artifact_contains( + &execution_root, + &definition, + Some(&FiniteReviewRuntimeContext { + candidate_identity: "candidate-under-test".to_string(), + acceptance_criteria: vec![ + "exact workflow validator".to_string(), + "persisted state transitions".to_string(), + "artifact verifier".to_string(), + ], + artifact: REVIEW_ARTIFACT.to_string(), + artifact_preexisting: true, + admitted_at_ms: 0, + }), + ) + .expect("preexisting finite review should evaluate deterministically"); + assert!( + !preexisting.passed, + "preexisting review artifacts must fail" + ); + + let stale_identity = evaluate_artifact_contains( + &execution_root, + &definition, + Some(&FiniteReviewRuntimeContext { + candidate_identity: "different-candidate".to_string(), + acceptance_criteria: vec![ + "exact workflow validator".to_string(), + "persisted state transitions".to_string(), + "artifact verifier".to_string(), + ], + artifact: REVIEW_ARTIFACT.to_string(), + artifact_preexisting: false, + admitted_at_ms: 0, + }), + ) + .expect("stale finite review should evaluate deterministically"); + assert!( + !stale_identity.passed, + "review artifacts must bind the admitted candidate identity" + ); +} + +#[tokio::test] +async fn synthetic_finance_acceptance_uses_real_runtime_and_zero_finance_boundaries() { + let candidate_sha = std::env::var("CODEWITH_CANDIDATE_SHA") + .unwrap_or_else(|_| "candidate-under-test".to_string()); + + let structural_yaml = synthetic_finance_workflow_yaml("wf_finance_structural_control").replace( + "title: \"Build the deterministic analysis artifact\"", + "title: \"Submit payment provider request\"", + ); + let structural_spec = codex_workflows::parse_workflow_yaml(&structural_yaml) + .expect("the negative structural fixture should pass the generic parser"); + let structural_error = validate_synthetic_finance_structure(&structural_spec) + .expect_err("a payment/provider execution step must be structurally rejected"); + assert!( + structural_error + .to_string() + .contains("payment/provider execution step") + ); + + let injected = SyntheticFinanceHarness::new("wf_finance_injected_boundary") + .await + .expect("injected-call harness should initialize"); + injected + .complete_candidate(&candidate_sha) + .await + .expect("injected-call candidate should pass"); + let mut injected_counters = FinanceBoundaryCounters::default(); + let injected_error = injected_counters + .inject_bank_provider_submission_attempt() + .expect_err("finance boundary must reject before provider I/O"); + assert!(injected_error.to_string().contains("before I/O")); + let injected_review = injected + .prepare_step( + REVIEW_STEP_ID, + REVIEW_ARTIFACT, + &review_artifact(&candidate_sha, Some("GO"), injected_counters.total()), + ) + .await + .expect("injected-call review should reach its verifier"); + assert!(!injected_review.evaluation.passed); + let injected_snapshot = injected + .execute_prepared(injected_review) + .await + .expect("injected-call verifier failure should persist"); + assert_eq!(WorkflowRunStatus::Failed, injected_snapshot.run.status); + + let missing_field = SyntheticFinanceHarness::new("wf_finance_missing_field") + .await + .expect("missing-field harness should initialize"); + missing_field + .complete_candidate(&candidate_sha) + .await + .expect("missing-field candidate should pass"); + let missing_review = missing_field + .prepare_step( + REVIEW_STEP_ID, + REVIEW_ARTIFACT, + &review_artifact( + &candidate_sha, + /*verdict*/ None, + /*finance_calls*/ 0, + ), + ) + .await + .expect("missing-field review should reach its verifier"); + assert!(!missing_review.evaluation.passed); + let missing_snapshot = missing_field + .execute_prepared(missing_review) + .await + .expect("missing-field verifier failure should persist"); + assert_eq!(WorkflowRunStatus::Failed, missing_snapshot.run.status); + + let no_go = SyntheticFinanceHarness::new("wf_finance_no_go") + .await + .expect("NO_GO harness should initialize"); + no_go + .complete_candidate(&candidate_sha) + .await + .expect("NO_GO candidate should pass"); + let no_go_artifact = review_artifact(&candidate_sha, Some("GO"), /*finance_calls*/ 0).replace( + "verdict: GO\nblocking_p0_p1: []", + "verdict: NO_GO\nblocking_p0_p1:\n - severity: P1\n finding: reachable blocker", + ); + let no_go_review = no_go + .prepare_step(REVIEW_STEP_ID, REVIEW_ARTIFACT, &no_go_artifact) + .await + .expect("NO_GO review should reach its verifier"); + assert!(!no_go_review.evaluation.passed); + let no_go_snapshot = no_go + .execute_prepared(no_go_review) + .await + .expect("NO_GO verifier failure should persist"); + assert_eq!(WorkflowRunStatus::Failed, no_go_snapshot.run.status); + + let blocking_p0_p1 = SyntheticFinanceHarness::new("wf_finance_blocking_p0_p1") + .await + .expect("blocking-P0/P1 harness should initialize"); + blocking_p0_p1 + .complete_candidate(&candidate_sha) + .await + .expect("blocking-P0/P1 candidate should pass"); + let blocking_p0_p1_artifact = + review_artifact(&candidate_sha, Some("GO"), /*finance_calls*/ 0).replace( + "blocking_p0_p1: []", + "blocking_p0_p1:\n - severity: P1\n finding: reachable blocker", + ); + let blocking_p0_p1_review = blocking_p0_p1 + .prepare_step(REVIEW_STEP_ID, REVIEW_ARTIFACT, &blocking_p0_p1_artifact) + .await + .expect("blocking-P0/P1 review should reach its verifier"); + assert!(!blocking_p0_p1_review.evaluation.passed); + let blocking_p0_p1_snapshot = blocking_p0_p1 + .execute_prepared(blocking_p0_p1_review) + .await + .expect("blocking-P0/P1 verifier failure should persist"); + assert_eq!( + WorkflowRunStatus::Failed, + blocking_p0_p1_snapshot.run.status + ); + + let remediation_over_cap = SyntheticFinanceHarness::new("wf_finance_remediation_over_cap") + .await + .expect("remediation-over-cap harness should initialize"); + remediation_over_cap + .complete_candidate(&candidate_sha) + .await + .expect("remediation-over-cap candidate should pass"); + let remediation_over_cap_artifact = + review_artifact(&candidate_sha, Some("GO"), /*finance_calls*/ 0).replace( + "remediation_cycle: 0\nremediation_cycle_cap: 2", + "remediation_cycle: 3\nremediation_cycle_cap: 2", + ); + let remediation_over_cap_review = remediation_over_cap + .prepare_step( + REVIEW_STEP_ID, + REVIEW_ARTIFACT, + &remediation_over_cap_artifact, + ) + .await + .expect("remediation-over-cap review should reach its verifier"); + assert!(!remediation_over_cap_review.evaluation.passed); + let remediation_over_cap_snapshot = remediation_over_cap + .execute_prepared(remediation_over_cap_review) + .await + .expect("remediation-over-cap verifier failure should persist"); + assert_eq!( + WorkflowRunStatus::Failed, + remediation_over_cap_snapshot.run.status + ); + + let success = SyntheticFinanceHarness::new("wf_finance_acceptance") + .await + .expect("success harness should initialize"); + success + .complete_candidate(&candidate_sha) + .await + .expect("success candidate should pass"); + let prepared_review = success + .prepare_step( + REVIEW_STEP_ID, + REVIEW_ARTIFACT, + &review_artifact(&candidate_sha, Some("GO"), /*finance_calls*/ 0), + ) + .await + .expect("success review should reach its verifier"); + assert!(prepared_review.evaluation.passed); + let missing_receipt = terminal_acceptance_receipt( + &success, + &prepared_review.snapshot, + &candidate_sha, + FinanceBoundaryCounters::default(), + &prepared_review.evaluation, + ) + .expect_err("a run without a verifier-gated terminal receipt must fail acceptance"); + assert!( + missing_receipt + .to_string() + .contains("terminal verifier-gated workflow receipt is missing") + ); + let review_evaluation = prepared_review.evaluation.clone(); + let terminal_snapshot = success + .execute_prepared(prepared_review) + .await + .expect("success review verifier should persist"); + let reopened = StateRuntime::init( + success.runtime.codex_home().to_path_buf(), + "test-provider-reopened".to_string(), + ) + .await + .expect("isolated workflow persistence should reopen"); + let persisted_snapshot = reopened + .workflows() + .get_workflow_run_snapshot(&success.run_id) + .await + .expect("persisted workflow snapshot should load") + .expect("persisted workflow run should exist"); + assert_eq!(terminal_snapshot.run.status, persisted_snapshot.run.status); + assert_eq!(terminal_snapshot.steps, persisted_snapshot.steps); + assert_eq!(terminal_snapshot.verifiers, persisted_snapshot.verifiers); + let receipt = terminal_acceptance_receipt( + &success, + &persisted_snapshot, + &candidate_sha, + FinanceBoundaryCounters::default(), + &review_evaluation, + ) + .expect("terminal acceptance receipt should validate"); + + let receipt_json = serde_json::to_string_pretty(&receipt).expect("receipt should serialize"); + println!("SYNTHETIC_FINANCE_ACCEPTANCE_RECEIPT={receipt_json}"); + if let Ok(path) = std::env::var("CODEWITH_FINANCE_ACCEPTANCE_RECEIPT") { + std::fs::write(path, format!("{receipt_json}\n")) + .expect("remote acceptance receipt should write"); + } +} diff --git a/codex-rs/prompts/src/workflows_tests.rs b/codex-rs/prompts/src/workflows_tests.rs index 48e6e1111a..6f6df8ed52 100644 --- a/codex-rs/prompts/src/workflows_tests.rs +++ b/codex-rs/prompts/src/workflows_tests.rs @@ -30,6 +30,34 @@ fn workflow_prompt_requires_deep_adversarial_verified_workflows() { } } +#[test] +fn workflow_prompt_requires_one_bounded_independent_review() { + for required in [ + "exactly one independent adversarial reviewer agent", + "review step assigned to that reviewer", + "exact candidate and the exact acceptance criteria", + "candidate_identity:", + "acceptance_criteria:", + "verdict:", + "blocking_p0_p1:", + "non_blocking_p2_p3:", + "remediation_cycle:", + "remediation_cycle_cap: 2", + "zero open blocking P0/P1 findings", + "at most two focused remediation cycles", + "A third `NO_GO` stops", + ] { + assert!( + WORKFLOW_YAML_SYSTEM_PROMPT.contains(required), + "missing bounded-review prompt fragment: {required}" + ); + } + assert!( + !WORKFLOW_YAML_SYSTEM_PROMPT.contains("at least two agents or steps"), + "the generator must not require duplicate adversarial review" + ); +} + #[test] fn workflow_prompt_requires_ancient_agent_names_and_model_routing() { for required in [ diff --git a/codex-rs/prompts/templates/workflows/deep_yaml_system_prompt.md b/codex-rs/prompts/templates/workflows/deep_yaml_system_prompt.md index 9ccb425bce..6ce33e1ce5 100644 --- a/codex-rs/prompts/templates/workflows/deep_yaml_system_prompt.md +++ b/codex-rs/prompts/templates/workflows/deep_yaml_system_prompt.md @@ -48,8 +48,12 @@ Completion and deterministic verification: - The workflow becomes complete only after every required step is `succeeded` and every workflow-level verifier passes. Adversarial and testing work: -- Every workflow must include adversarial work by at least two agents or steps. These reviewers should challenge scope, security, correctness, data quality, UX, cost, and operational assumptions. +- Every workflow must include adversarial work through exactly one independent adversarial reviewer agent and exactly one initial review step assigned to that reviewer. Use `review` or `adversarial` in the review step's stable id. The review step must depend on a candidate-producing step owned by a different agent. Do not add a second reviewer or reviewer-per-remediation steps. - Adversarial review is a required workflow artifact, not optional guidance. +- The review step must inspect the exact candidate and the exact acceptance criteria, then emit one `GO` or `NO_GO` verdict. `GO` requires all applicable gates to pass with zero open blocking P0/P1 findings. `NO_GO` must name every blocking P0/P1 defect and its evidence. +- Only concrete, evidence-backed, currently reachable, in-scope P0/P1 defects material to acceptance, secrets or security, data or session integrity, unsafe mutation or rollback, or an applicable required gate may block. Record P2/P3, speculative, pre-existing, and out-of-scope findings once as non-blocking follow-ups. +- The required review artifact must expose `candidate_identity:`, `acceptance_criteria:`, `verdict:`, `blocking_p0_p1:`, `non_blocking_p2_p3:`, `remediation_cycle:`, and `remediation_cycle_cap: 2`. Keep it in the review step's `outputs` and top-level `artifacts.required`, and verifier-gate those exact fields with an `artifact_contains` verifier. +- The same reviewer may perform at most two focused remediation cycles over named blocking defects and direct regressions. A third `NO_GO` stops and reports the remaining blockers; it does not start a third fix cycle. - Include negative cases, boundary cases, failure-mode review, and attempts to disprove completion claims. - Every implementation or launch path must include deterministic verification and test evidence. - Reviews are not sufficient evidence by themselves; include machine-checkable tests, scripts, fixtures, audits, or acceptance gates where possible. diff --git a/codex-rs/state/src/runtime/workflow_orchestrator.rs b/codex-rs/state/src/runtime/workflow_orchestrator.rs index 3631c7d2a1..51bbd1faf1 100644 --- a/codex-rs/state/src/runtime/workflow_orchestrator.rs +++ b/codex-rs/state/src/runtime/workflow_orchestrator.rs @@ -23,11 +23,15 @@ use codex_git_utils::remove_linked_git_worktree; use codex_git_utils::resolve_git_ref; use codex_protocol::models::PermissionProfile; use codex_workflows::WorkflowBranchPrompt; +use codex_workflows::WorkflowVerifier; use codex_workflows::WorkflowWorkspace; use codex_workflows::WorkflowWorkspaceMode; use codex_workflows::render_workflow_branch_prompt; +use codex_workflows::verifier_has_finite_review_artifact_contract; use serde_json::Value; use serde_json::json; +use sha2::Digest; +use sha2::Sha256; use sqlx::Row; use std::path::Path; use std::path::PathBuf; @@ -704,12 +708,16 @@ async fn admit_ready_workflow_branches_in_tx( existing_background_agent_run_id_by_idempotency_key_in_tx(tx, idempotency_key.as_str()) .await? .unwrap_or_else(|| Uuid::new_v4().to_string()); + let review_basis = workflow_review_branch_basis_in_tx(tx, run, &candidate).await?; let provisioned_workspace = provision_workflow_workspace( codex_home, run, &candidate, workspace_mode, branch_attempt, + review_basis + .as_ref() + .map(|basis| basis.candidate_identity.as_str()), )?; if provisioned_workspace.created_linked_worktree { let Some(branch) = provisioned_workspace.branch.clone() else { @@ -725,6 +733,8 @@ async fn admit_ready_workflow_branches_in_tx( branch, }); } + let review_context = + workflow_review_branch_context(review_basis.as_ref(), &provisioned_workspace)?; let admission_json = workflow_state_json_string( "workflow_branch_admission", json!({ @@ -738,6 +748,8 @@ async fn admit_ready_workflow_branches_in_tx( "workspace": workspace_json, "managedWorktreeId": provisioned_workspace.worktree_id.as_str(), "cwd": provisioned_workspace.execution_cwd, + "headSha": provisioned_workspace.head_sha, + "reviewContext": review_context.as_ref().map(|context| &context.state_json), }), )?; let updated = sqlx::query( @@ -786,6 +798,7 @@ WHERE step_run_id = ? model_route_json: &model_route_json, workspace_json: workspace_json.as_ref(), provisioned_workspace: &provisioned_workspace, + review_context: review_context.as_ref(), background_agent_run_id: background_agent_run_id.as_str(), idempotency_key: idempotency_key.as_str(), params, @@ -1067,12 +1080,25 @@ struct BackgroundBranchRunCreate<'a> { model_route_json: &'a Value, workspace_json: Option<&'a Value>, provisioned_workspace: &'a ProvisionedWorkflowWorkspace, + review_context: Option<&'a WorkflowReviewBranchContext>, background_agent_run_id: &'a str, idempotency_key: &'a str, params: &'a WorkflowRunBranchAdmissionParams, now_ms: i64, } +struct WorkflowReviewBranchContext { + prompt_suffix: String, + state_json: Value, +} + +struct WorkflowReviewBranchBasis { + verifier: WorkflowVerifier, + candidate_identity: String, + acceptance_criteria: Vec, + dependency_evidence: Vec, +} + struct ProvisionedWorkflowWorkspace { worktree_id: String, mode: WorkflowWorkspaceMode, @@ -1319,19 +1345,24 @@ async fn create_background_branch_run_if_missing_in_tx( model_route_json, workspace_json, provisioned_workspace, + review_context, background_agent_run_id, idempotency_key, params, now_ms, } = branch; let now = now_ms.div_euclid(1000); - let prompt = render_workflow_branch_prompt(WorkflowBranchPrompt { + let mut prompt = render_workflow_branch_prompt(WorkflowBranchPrompt { run_id: run.run_id.as_str(), step_id: candidate.step_id.as_str(), title: candidate.title.as_str(), agent_id: candidate.agent_id.as_str(), parallel_group: candidate.parallel_group.as_deref(), }); + if let Some(review_context) = review_context { + prompt.push_str("\n\n"); + prompt.push_str(review_context.prompt_suffix.as_str()); + } let prompt_snapshot_ref = format!("workflow:{}:step:{}:prompt", run.run_id, candidate.step_id); let spawn_linkage_json = json!({ "schemaVersion": "workflow.branch_spawn/v0", @@ -1382,8 +1413,8 @@ async fn create_background_branch_run_if_missing_in_tx( model_route_json, workspace_json, provisioned_workspace, + review_context, params, - recovery_policy, ), recovery_policy: recovery_policy.to_string(), config_fingerprint: params.config_fingerprint.clone(), @@ -1519,14 +1550,271 @@ ON CONFLICT(run_id) DO UPDATE SET Ok(()) } +async fn workflow_review_branch_basis_in_tx( + tx: &mut sqlx::Transaction<'_, Sqlite>, + run: &crate::WorkflowRun, + candidate: &ReadyBranchCandidate, +) -> anyhow::Result> { + let verifier_rows = sqlx::query( + r#" +SELECT definition_json +FROM workflow_run_step_verifiers +WHERE run_id = ? AND step_id = ? AND verifier_type = 'artifact_contains' +ORDER BY verifier_id + "#, + ) + .bind(run.run_id.as_str()) + .bind(candidate.step_id.as_str()) + .fetch_all(&mut **tx) + .await?; + let mut review_verifier = None; + for row in verifier_rows { + let definition_json: String = row.try_get("definition_json")?; + let definition_json: Value = serde_json::from_str(definition_json.as_str())?; + let definition: WorkflowVerifier = + serde_json::from_value(workflow_state_data(&definition_json).clone())?; + if verifier_has_finite_review_artifact_contract(&definition) { + review_verifier = Some(definition); + break; + } + } + let Some(review_verifier) = review_verifier else { + return Ok(None); + }; + let dependency_rows = sqlx::query( + r#" +SELECT + dependency.depends_on_step_id, + step.title, + step.agent_id, + step.status, + step.background_agent_run_id, + step.branch_admission_json, + step.completion_model_marked_state, + step.completed_at_ms +FROM workflow_run_step_dependencies dependency +JOIN workflow_run_steps step + ON step.run_id = dependency.run_id + AND step.step_id = dependency.depends_on_step_id +WHERE dependency.run_id = ? AND dependency.step_id = ? +ORDER BY dependency.depends_on_step_id + "#, + ) + .bind(run.run_id.as_str()) + .bind(candidate.step_id.as_str()) + .fetch_all(&mut **tx) + .await?; + if dependency_rows.is_empty() { + anyhow::bail!("finite review step must depend on an exact candidate step"); + } + let mut dependency_evidence = Vec::with_capacity(dependency_rows.len()); + let mut acceptance_criteria = Vec::new(); + for row in dependency_rows { + let step_id: String = row.try_get("depends_on_step_id")?; + let status: String = row.try_get("status")?; + if status != crate::WorkflowRunStepStatus::Succeeded.as_str() { + anyhow::bail!("finite review dependency `{step_id}` has not succeeded"); + } + let branch_admission_json = row + .try_get::, _>("branch_admission_json")? + .ok_or_else(|| anyhow::anyhow!("review dependency has no branch admission"))?; + let branch_admission_json: Value = serde_json::from_str(branch_admission_json.as_str())?; + let dependency_admission = workflow_state_data(&branch_admission_json); + let dependency_cwd = dependency_admission + .get("cwd") + .and_then(Value::as_str) + .map(PathBuf::from) + .ok_or_else(|| anyhow::anyhow!("review dependency has no persisted cwd"))?; + let candidate_status = get_git_worktree_status_snapshot(dependency_cwd.as_path())?; + if candidate_status.dirty { + anyhow::bail!( + "finite review dependency `{step_id}` has uncommitted changes and no exact reviewable identity" + ); + } + let candidate_head_sha = candidate_status + .head_sha + .as_deref() + .filter(|value| !value.is_empty()) + .ok_or_else(|| anyhow::anyhow!("review dependency has no exact HEAD identity"))?; + let verifier_rows = sqlx::query( + r#" +SELECT verifier_id, verifier_type, status, definition_json, last_result_json +FROM workflow_run_step_verifiers +WHERE run_id = ? AND step_id = ? +ORDER BY verifier_id + "#, + ) + .bind(run.run_id.as_str()) + .bind(step_id.as_str()) + .fetch_all(&mut **tx) + .await?; + if verifier_rows.is_empty() { + anyhow::bail!( + "finite review dependency `{step_id}` has no deterministic verifier evidence" + ); + } + let mut verifier_evidence = Vec::with_capacity(verifier_rows.len()); + for verifier_row in verifier_rows { + let verifier_id: String = verifier_row.try_get("verifier_id")?; + let verifier_type: String = verifier_row.try_get("verifier_type")?; + let verifier_status: String = verifier_row.try_get("status")?; + if verifier_status != crate::WorkflowRunStepVerifierStatus::Passed.as_str() { + anyhow::bail!( + "finite review dependency `{step_id}` verifier `{verifier_id}` has not passed" + ); + } + let definition_json: String = verifier_row.try_get("definition_json")?; + let definition_json: Value = serde_json::from_str(definition_json.as_str())?; + let definition: WorkflowVerifier = + serde_json::from_value(workflow_state_data(&definition_json).clone())?; + let result_json = verifier_row + .try_get::, _>("last_result_json")? + .ok_or_else(|| { + anyhow::anyhow!( + "finite review dependency `{step_id}` verifier `{verifier_id}` has no result evidence" + ) + })?; + let result_json: Value = serde_json::from_str(result_json.as_str())?; + let result = workflow_state_data(&result_json).clone(); + let definition_canonical = serde_json::to_string(&definition)?; + let result_canonical = serde_json::to_string(&result)?; + let definition_sha256 = + format!("{:x}", Sha256::digest(definition_canonical.as_bytes())); + let result_sha256 = format!("{:x}", Sha256::digest(result_canonical.as_bytes())); + acceptance_criteria.push(format!( + "dependency `{step_id}` verifier `{verifier_id}` ({verifier_type}) contract: {definition_canonical}" + )); + let outcome = json!({ + "status": result.get("status"), + "expectedExitCode": result.get("expectedExitCode"), + "observedExitCode": result.get("observedExitCode"), + "timedOut": result.get("timedOut"), + "outputTruncated": result.get("outputTruncated"), + }); + acceptance_criteria.push(format!( + "dependency `{step_id}` verifier `{verifier_id}` observed outcome: {}", + serde_json::to_string(&outcome)? + )); + verifier_evidence.push(json!({ + "verifierId": verifier_id, + "verifierType": verifier_type, + "status": verifier_status, + "definition": definition, + "definitionSha256": definition_sha256, + "result": result, + "resultSha256": result_sha256, + })); + } + acceptance_criteria.push(format!( + "dependency `{step_id}` completed as `{status}` at exact HEAD `{candidate_head_sha}`" + )); + dependency_evidence.push(json!({ + "stepId": step_id, + "title": row.try_get::("title")?, + "agentId": row.try_get::("agent_id")?, + "status": status, + "completionModelMarkedState": row.try_get::, _>("completion_model_marked_state")?, + "backgroundAgentRunId": row.try_get::, _>("background_agent_run_id")?, + "candidateHeadSha": candidate_head_sha, + "candidateDirty": candidate_status.dirty, + "completedAtMs": row.try_get::, _>("completed_at_ms")?, + "verifiers": verifier_evidence, + })); + } + let candidate_identities = dependency_evidence + .iter() + .filter_map(|evidence| evidence.get("candidateHeadSha").and_then(Value::as_str)) + .collect::>(); + if candidate_identities.len() != 1 { + anyhow::bail!("finite review dependencies do not share one exact candidate HEAD identity"); + } + let candidate_identity = candidate_identities + .into_iter() + .next() + .ok_or_else(|| anyhow::anyhow!("finite review branch has no candidate identity"))? + .to_string(); + Ok(Some(WorkflowReviewBranchBasis { + verifier: review_verifier, + candidate_identity, + acceptance_criteria, + dependency_evidence, + })) +} + +fn workflow_review_branch_context( + basis: Option<&WorkflowReviewBranchBasis>, + provisioned_workspace: &ProvisionedWorkflowWorkspace, +) -> anyhow::Result> { + let Some(basis) = basis else { + return Ok(None); + }; + let review_artifact = basis + .verifier + .artifact + .as_deref() + .ok_or_else(|| anyhow::anyhow!("finite review verifier has no artifact"))?; + let review_artifact_path = Path::new(review_artifact); + if review_artifact_path.is_absolute() + || review_artifact_path + .components() + .any(|component| !matches!(component, std::path::Component::Normal(_))) + { + anyhow::bail!("finite review artifact must be a normalized workspace-relative path"); + } + let artifact_preexisting = match std::fs::symlink_metadata( + provisioned_workspace + .execution_cwd + .join(review_artifact_path), + ) { + Ok(_) => true, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => false, + Err(error) => return Err(error.into()), + }; + let required_fields = basis + .verifier + .must_contain + .iter() + .map(|field| field.trim().to_string()) + .collect::>(); + let dependency_json = serde_json::to_string(&basis.dependency_evidence)?; + let required_fields_text = required_fields + .iter() + .map(|field| format!("- {field}")) + .collect::>() + .join("\n"); + let candidate_identity = basis.candidate_identity.as_str(); + let prompt_suffix = format!( + "Independent review contract for this exact workflow candidate:\n\ + Candidate identity: {candidate_identity}\n\ + Acceptance criteria: {}\n\ + Dependency evidence: {dependency_json}\n\ + Create a new YAML review artifact at `{review_artifact}`. A file that existed before this reviewer started is invalid.\n\ + The artifact must contain these fields:\n{required_fields_text}\n\ + Set `verdict: GO` only when `blocking_p0_p1` is an empty list. `NO_GO`, a non-empty blocking list, a different candidate identity, or a remediation cycle outside 0..2 fails the deterministic gate.", + basis.acceptance_criteria.join("; ") + ); + Ok(Some(WorkflowReviewBranchContext { + prompt_suffix, + state_json: json!({ + "schemaVersion": "workflow.review_context/v0", + "candidateIdentity": basis.candidate_identity.as_str(), + "acceptanceCriteria": &basis.acceptance_criteria, + "dependencyEvidence": &basis.dependency_evidence, + "artifact": review_artifact, + "artifactPreexisting": artifact_preexisting, + "requiredFields": required_fields, + }), + })) +} + fn branch_execution_payload( run: &crate::WorkflowRun, candidate: &ReadyBranchCandidate, model_route_json: &Value, workspace_json: Option<&Value>, provisioned_workspace: &ProvisionedWorkflowWorkspace, + review_context: Option<&WorkflowReviewBranchContext>, params: &WorkflowRunBranchAdmissionParams, - recovery_policy: &str, ) -> Value { json!({ "snapshotSource": "workflow/branch_admission", @@ -1549,11 +1837,12 @@ fn branch_execution_payload( .as_deref() .map(|profile| StateRuntime::background_agent_identity_sha256(profile.as_bytes())), "workspace": workspace_json, + "reviewContext": review_context.map(|context| &context.state_json), "envSnapshotPolicy": "inherit-minimal", "configFingerprint": params.config_fingerprint, "versionFingerprint": params.version_fingerprint, "packageFingerprint": params.runtime_package_fingerprint, - "recoveryPolicy": recovery_policy, + "recoveryPolicy": WORKFLOW_BRANCH_RECOVERY_POLICY, "maxRuntimeSeconds": workflow_state_data(&run.limits_json).get("max_step_runtime_seconds"), }) } @@ -1647,6 +1936,7 @@ fn provision_workflow_workspace( candidate: &ReadyBranchCandidate, mode: WorkflowWorkspaceMode, attempt: i64, + exact_start_point: Option<&str>, ) -> anyhow::Result { let base_repo_path = run.source_repo_path.clone().ok_or_else(|| { anyhow::anyhow!( @@ -1681,6 +1971,14 @@ fn provision_workflow_workspace( match mode { WorkflowWorkspaceMode::SharedRepository => { let status_snapshot = get_git_worktree_status_snapshot(base_repo_path.as_path())?; + if let Some(exact_start_point) = exact_start_point + && (status_snapshot.dirty + || status_snapshot.head_sha.as_deref() != Some(exact_start_point)) + { + anyhow::bail!( + "shared review workspace does not match exact candidate HEAD `{exact_start_point}`" + ); + } Ok(ProvisionedWorkflowWorkspace { worktree_id, mode, @@ -1695,13 +1993,19 @@ fn provision_workflow_workspace( }) } WorkflowWorkspaceMode::IsolatedWorktree => { - let start_point = - resolve_git_ref(base_repo_path.as_path(), "HEAD")?.ok_or_else(|| { + let requested_start_point = exact_start_point.unwrap_or("HEAD"); + let start_point = resolve_git_ref(base_repo_path.as_path(), requested_start_point)? + .ok_or_else(|| { anyhow::anyhow!( - "workflow source repository {} has no resolvable HEAD", - base_repo_path.display() + "workflow source repository {} cannot resolve start point `{requested_start_point}`", + base_repo_path.display(), ) })?; + if exact_start_point.is_some_and(|expected| start_point != expected) { + anyhow::bail!( + "workflow review start point `{requested_start_point}` resolved to unexpected HEAD `{start_point}`" + ); + } let repo_key = StateRuntime::background_agent_identity_sha256( base_repo_path.to_string_lossy().as_bytes(), ); @@ -2875,16 +3179,28 @@ cleanup: "Ptolemy", "Aquinas", ]; - let agents = (0..step_count) + let mut agents = (0..step_count) .map(|index| { - let display_name = format!( - "Adversary-{}", - ancient_names.get(index).copied().unwrap_or("Aristotle") - ); + let display_name = if step_count == 1 { + format!( + "Builder-{}", + ancient_names.get(index).copied().unwrap_or("Aristotle") + ) + } else { + format!( + "Adversary-{}", + ancient_names.get(index).copied().unwrap_or("Aristotle") + ) + }; + let role = if step_count == 1 { + format!("Exercise branch {index} admission.") + } else { + format!("Review branch {index}.") + }; format!( r#" - id: "agent_{index}" display_name: "{display_name}" - role: "Review branch {index}." + role: "{role}" model: model_gateway: "hasna" provider: "openai" @@ -2894,7 +3210,7 @@ cleanup: ) }) .collect::(); - let steps = (0..step_count) + let mut steps = (0..step_count) .map(|index| { let route = if index == 0 { r#" model: @@ -2937,6 +3253,53 @@ cleanup: ) }) .collect::(); + let required_artifacts = if step_count == 1 { + agents.push_str( + r#" - id: "adversarial_reviewer" + display_name: "Reviewer-Seneca" + role: "Independently review the exact branch admission candidate." + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" +"#, + ); + steps.push_str( + r#" - id: "initial_adversarial_review" + title: "Run the initial adversarial review" + agent: "adversarial_reviewer" + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" + depends_on: + - "branch_0" + outputs: + - "review.yaml" + completion: + model_marked_state: "candidate_succeeded" + verifiers: + - id: "finite_review_artifact_contract" + type: "artifact_contains" + artifact: "review.yaml" + must_contain: + - "candidate_identity:" + - "acceptance_criteria:" + - "verdict:" + - "blocking_p0_p1:" + - "non_blocking_p2_p3:" + - "remediation_cycle:" + - "remediation_cycle_cap: 2" +"#, + ); + r#" + - "review.yaml" +"# + } else { + "[]\n" + }; format!( r#"schema_version: "workflow.codex.codewith/v0" workflow_id: "{workflow_id}" @@ -2962,8 +3325,7 @@ agents: {agents}steps: {steps}artifacts: retention: "until_workflow_complete" - required: [] -cleanup: + required: {required_artifacts}cleanup: on_cancel: [] on_complete: [] "# @@ -3729,7 +4091,7 @@ WHERE worktree_id = ? "wf_branch_typed_permission_profile", /*step_count*/ 1, /*max_parallel_steps*/ 1, - /*max_agents*/ 1, + /*max_agents*/ 2, /*max_worktrees*/ 1, "typed-permission-profile", ), @@ -3796,7 +4158,7 @@ WHERE worktree_id = ? "wf_branch_invalid_permission_profile", /*step_count*/ 1, /*max_parallel_steps*/ 1, - /*max_agents*/ 1, + /*max_agents*/ 2, /*max_worktrees*/ 1, "invalid-permission-profile", ), @@ -3884,7 +4246,7 @@ WHERE worktree_id = ? "wf_branch_provision_rollback", /*step_count*/ 1, /*max_parallel_steps*/ 1, - /*max_agents*/ 1, + /*max_agents*/ 2, /*max_worktrees*/ 1, "provision-rollback", ), diff --git a/codex-rs/workflows/src/lib.rs b/codex-rs/workflows/src/lib.rs index 10503065c6..8295998745 100644 --- a/codex-rs/workflows/src/lib.rs +++ b/codex-rs/workflows/src/lib.rs @@ -46,6 +46,7 @@ pub use spec::WorkflowVerifier; pub use spec::WorkflowVerifierRetryPolicy; pub use spec::WorkflowWorkspace; pub use spec::WorkflowWorkspaceMode; +pub use validation::verifier_has_finite_review_artifact_contract; pub const MAX_WORKFLOW_PROMPT_FIELD_CHARS: usize = 240; pub const MAX_WORKFLOW_YAML_BYTES: usize = 256 * 1024; diff --git a/codex-rs/workflows/src/tests.rs b/codex-rs/workflows/src/tests.rs index c232a706ce..e93f8bb7c1 100644 --- a/codex-rs/workflows/src/tests.rs +++ b/codex-rs/workflows/src/tests.rs @@ -19,7 +19,7 @@ const SINGLE_REVIEWER_WORKFLOW_YAML: &str = r#" schema_version: "workflow.codex.codewith/v0" workflow_id: "wf_single_adversarial_review" display_name: "Single adversarial review" -source_prompt: "Run one independent adversarial review." +source_prompt: "Build one candidate and run one independent adversarial review." status: "draft" execution_defaults: model_gateway: "hasna" @@ -28,7 +28,7 @@ execution_defaults: reasoning: "high" limits: max_parallel_steps: 1 - max_agents: 1 + max_agents: 2 max_worktrees: 1 max_runtime_seconds: 3600 max_step_runtime_seconds: 1200 @@ -37,15 +37,42 @@ limits: approvals: required_before: [] agents: + - id: "candidate_builder" + display_name: "Builder-Vitruvius" + role: "Build the exact candidate and record its acceptance criteria." + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" - id: "adversarial_reviewer" - display_name: "Adversary-Hypatia" - role: "Run the one independent adversarial review." + display_name: "Reviewer-Hypatia" + role: "Independently review the exact candidate against its acceptance criteria." model: model_gateway: "hasna" provider: "openai" model: "gpt-5.4" reasoning: "high" steps: + - id: "build_candidate" + title: "Build the exact candidate" + agent: "candidate_builder" + model: + model_gateway: "hasna" + provider: "openai" + model: "gpt-5.4" + reasoning: "high" + depends_on: [] + outputs: + - "candidate.txt" + completion: + model_marked_state: "candidate_succeeded" + verifiers: + - id: "candidate_identity_present" + type: "artifact_contains" + artifact: "candidate.txt" + must_contain: + - "candidate_identity:" - id: "initial_adversarial_review" title: "Run the initial adversarial review" agent: "adversarial_reviewer" @@ -54,21 +81,29 @@ steps: provider: "openai" model: "gpt-5.4" reasoning: "high" - depends_on: [] + depends_on: + - "build_candidate" outputs: - - "review.md" + - "review.yaml" completion: model_marked_state: "candidate_succeeded" verifiers: - - id: "review_verdict_present" + - id: "finite_review_artifact_contract" type: "artifact_contains" - artifact: "review.md" + artifact: "review.yaml" must_contain: - - "GO" + - "candidate_identity:" + - "acceptance_criteria:" + - "verdict:" + - "blocking_p0_p1:" + - "non_blocking_p2_p3:" + - "remediation_cycle:" + - "remediation_cycle_cap: 2" artifacts: retention: "preserve_evidence" required: - - "review.md" + - "candidate.txt" + - "review.yaml" cleanup: on_cancel: - "stop_child_agents" @@ -353,14 +388,93 @@ fn parses_typed_workspace_modes_and_rejects_unknown_modes() { } #[test] -fn accepts_single_adversarial_reviewer_and_initial_review_step() { +fn accepts_one_independent_adversarial_review_run() { let spec = parse_workflow_yaml(SINGLE_REVIEWER_WORKFLOW_YAML) - .expect("one adversarial reviewer and one initial review step should parse"); + .expect("one independent reviewer-agent/review-step pair should parse"); - assert_eq!(1, spec.agents.len()); - assert_eq!(1, spec.steps.len()); - assert_eq!("adversarial_reviewer", spec.agents[0].id); - assert_eq!("initial_adversarial_review", spec.steps[0].id); + assert_eq!(2, spec.agents.len()); + assert_eq!(2, spec.steps.len()); + assert_eq!("candidate_builder", spec.agents[0].id); + assert_eq!("adversarial_reviewer", spec.agents[1].id); + assert_eq!("build_candidate", spec.steps[0].id); + assert_eq!("initial_adversarial_review", spec.steps[1].id); + assert_ne!(spec.steps[0].agent, spec.steps[1].agent); +} + +#[test] +fn accepts_legacy_multi_reviewer_workflow_without_paired_review_step() { + let yaml = SINGLE_REVIEWER_WORKFLOW_YAML + .replace("candidate_builder", "security_adversary") + .replace("Builder-Vitruvius", "Adversary-Cicero") + .replace( + "Build the exact candidate and record its acceptance criteria.", + "Challenge the candidate security assumptions.", + ) + .replace("initial_adversarial_review", "final_quality_audit") + .replace( + "Run the initial adversarial review", + "Run the final quality audit", + ); + + let spec = parse_workflow_yaml(&yaml) + .expect("legacy workflows with multiple adversarial agents must remain valid"); + + assert_eq!(2, spec.agents.len()); + assert_eq!(2, spec.steps.len()); +} + +#[test] +fn validates_synthetic_finance_retry_without_payment_execution_steps() { + let yaml = SINGLE_REVIEWER_WORKFLOW_YAML + .replace("wf_single_adversarial_review", "wf_synthetic_finance_retry") + .replace("Single adversarial review", "Synthetic finance retry") + .replace( + "Build one candidate and run one independent adversarial review.", + "Analyze synthetic invoice metadata and review the bounded evidence without execution.", + ) + .replace("build_candidate", "analyze_synthetic_invoice") + .replace( + "Build the exact candidate", + "Produce the synthetic finance analysis artifact", + ) + .replace("candidate.txt", "finance-analysis.txt"); + + let spec = parse_workflow_yaml(&yaml) + .expect("the non-executing synthetic finance workflow should validate"); + let reviewer_agents = spec + .agents + .iter() + .filter(|agent| agent.id == "adversarial_reviewer") + .count(); + let review_steps = spec + .steps + .iter() + .filter(|step| step.id == "initial_adversarial_review") + .count(); + let forbidden_execution_terms = [ + "pay", "approve", "schedule", "transfer", "submit", "bank", "provider", "mutate", + ]; + let payment_or_provider_execution_steps = spec + .steps + .iter() + .filter(|step| { + let identity = format!("{} {}", step.id, step.title).to_ascii_lowercase(); + forbidden_execution_terms.iter().any(|term| { + identity + .split(|ch: char| !ch.is_ascii_alphanumeric()) + .any(|part| part == *term) + }) + }) + .count(); + + assert_eq!(1, reviewer_agents); + assert_eq!(1, review_steps); + assert_eq!(0, payment_or_provider_execution_steps); + assert!(spec.steps.iter().all(|step| { + step.completion + .as_ref() + .is_some_and(|completion| !completion.verifiers.is_empty()) + })); } #[test] @@ -393,34 +507,115 @@ fn accepts_single_adversarial_reviewer_with_focused_rereview() { let spec = parse_workflow_yaml(&yaml) .expect("the same reviewer should be allowed to perform a focused re-review"); - assert_eq!(1, spec.agents.len()); - assert_eq!(2, spec.steps.len()); + assert_eq!(2, spec.agents.len()); + assert_eq!(3, spec.steps.len()); assert_eq!( - "adversarial_reviewer", spec.steps[1].agent, + "adversarial_reviewer", spec.steps[2].agent, "the focused re-review must reuse the fixed reviewer" ); } #[test] -fn rejects_draft_without_adversarial_reviewer_or_review_step() { +fn rejects_draft_without_adversarial_review_run() { let yaml = SINGLE_REVIEWER_WORKFLOW_YAML - .replace("adversarial_reviewer", "reviewer") - .replace("Adversary-Hypatia", "Reviewer-Hypatia") + .replace("adversarial_reviewer", "quality_auditor") + .replace("Reviewer-Hypatia", "Verifier-Hypatia") .replace( - "Run the one independent adversarial review.", - "Run the independent review.", + "Independently review the exact candidate against its acceptance criteria.", + "Audit the exact candidate against its acceptance criteria.", ) - .replace("initial_adversarial_review", "initial_review") + .replace("initial_adversarial_review", "initial_quality_audit") .replace( "Run the initial adversarial review", - "Run the initial review", + "Run the candidate audit", ); - let err = parse_workflow_yaml(&yaml).expect_err("adversarial work should remain required"); + let err = parse_workflow_yaml(&yaml) + .expect_err("a workflow without a reviewer-agent/review-step pair must fail"); assert_eq!( - "workflow spec is invalid: draft workflows must include adversarial work by at least one agent or one step", + "workflow spec is invalid: draft workflows must include at least one independent adversarial review run assigned to a reviewer agent", + err.to_string() + ); +} + +#[test] +fn rejects_unpaired_adversarial_reviewer() { + let yaml = SINGLE_REVIEWER_WORKFLOW_YAML + .replace("initial_adversarial_review", "final_quality_audit") + .replace( + "Run the initial adversarial review", + "Run the final quality audit", + ); + + let err = parse_workflow_yaml(&yaml) + .expect_err("declaring a reviewer without a review step is not a review run"); + + assert!( + err.to_string() + .contains("independent adversarial review run assigned to a reviewer agent"), + "unexpected error: {err}" + ); +} + +#[test] +fn rejects_adversarial_step_owned_by_non_reviewer() { + let yaml = SINGLE_REVIEWER_WORKFLOW_YAML + .replace("adversarial_reviewer", "quality_auditor") + .replace("Reviewer-Hypatia", "Verifier-Hypatia") + .replace( + "Independently review the exact candidate against its acceptance criteria.", + "Audit the exact candidate against its acceptance criteria.", + ); + + let err = parse_workflow_yaml(&yaml) + .expect_err("a review-named step assigned to a non-reviewer is not independent review"); + + assert!( err.to_string() + .contains("adversarial review step `initial_adversarial_review` must be assigned to a reviewer agent"), + "unexpected error: {err}" + ); +} + +#[test] +fn rejects_adversarial_review_without_verifier() { + let verifier = r#" verifiers: + - id: "finite_review_artifact_contract" + type: "artifact_contains" + artifact: "review.yaml" + must_contain: + - "candidate_identity:" + - "acceptance_criteria:" + - "verdict:" + - "blocking_p0_p1:" + - "non_blocking_p2_p3:" + - "remediation_cycle:" + - "remediation_cycle_cap: 2" +"#; + let yaml = SINGLE_REVIEWER_WORKFLOW_YAML.replace(verifier, " verifiers: []\n"); + + let err = parse_workflow_yaml(&yaml) + .expect_err("the existing completion verifier gate must reject the review step"); + + assert!( + err.to_string() + .contains("step `initial_adversarial_review` must include at least one verifier"), + "unexpected error: {err}" + ); +} + +#[test] +fn rejects_adversarial_review_without_finite_artifact_contract() { + let yaml = + SINGLE_REVIEWER_WORKFLOW_YAML.replace("remediation_cycle_cap: 2", "review_cycle_limit: 2"); + + let err = parse_workflow_yaml(&yaml) + .expect_err("a one-review workflow must deterministically verify its artifact contract"); + + assert!( + err.to_string().contains("finite review artifact contract"), + "unexpected error: {err}" ); } diff --git a/codex-rs/workflows/src/validation.rs b/codex-rs/workflows/src/validation.rs index af57221ef2..f53ff76803 100644 --- a/codex-rs/workflows/src/validation.rs +++ b/codex-rs/workflows/src/validation.rs @@ -29,6 +29,17 @@ use crate::ancient_names::is_ancient_display_name; const CANDIDATE_SUCCEEDED: &str = "candidate_succeeded"; const ARTIFACT_CONTAINS_VERIFIER: &str = "artifact_contains"; const RUN_COMMANDS_VERIFIER: &str = "run_commands"; +// The v0 schema has no typed review-policy fields, so single-review workflows use the +// existing deterministic artifact verifier as their finite compatibility contract. +const REVIEW_ARTIFACT_CONTRACT_FIELDS: &[&str] = &[ + "candidate_identity:", + "acceptance_criteria:", + "verdict:", + "blocking_p0_p1:", + "non_blocking_p2_p3:", + "remediation_cycle:", + "remediation_cycle_cap: 2", +]; const MAX_VERIFIER_RETRY_ATTEMPTS: u32 = 5; const MAX_WORKFLOW_LOOPS: usize = 32; const MAX_WORKFLOW_LOOP_ITERATIONS: u32 = 10_000; @@ -759,28 +770,117 @@ fn validate_acyclic_dependencies(steps: &[WorkflowStep]) -> WorkflowSpecResult<( } fn validate_adversarial_work(spec: &WorkflowSpec) -> WorkflowSpecResult<()> { - let adversarial_agents = spec + let reviewer_agent_ids = spec .agents .iter() .filter(|agent| { - is_adversarial_text(&agent.id) - || is_adversarial_text(&agent.display_name) + is_reviewer_text(&agent.id) + || is_reviewer_text(&agent.display_name) || is_adversarial_text(&agent.role) }) - .count(); - let adversarial_steps = spec + .map(|agent| agent.id.as_str()) + .collect::>(); + + // Preserve the v0 compatibility contract for workflows that already model review + // across multiple adversarial agents. The stricter paired topology and finite artifact + // contract apply to the new single-review shape generated by the workflow prompt. + if reviewer_agent_ids.len() > 1 { + return Ok(()); + } + + let review_steps = spec .steps .iter() - .filter(|step| is_adversarial_text(&step.id) || is_adversarial_text(&step.title)) - .count(); - if adversarial_agents == 0 && adversarial_steps == 0 { + .filter(|step| is_review_step(step)) + .collect::>(); + + if let Some(step) = review_steps + .iter() + .find(|step| !reviewer_agent_ids.contains(step.agent.as_str())) + { + return Err(WorkflowSpecError::invalid(format!( + "adversarial review step `{}` must be assigned to a reviewer agent", + step.id + ))); + } + + let paired_review_steps = review_steps + .iter() + .copied() + .filter(|review_step| { + review_step.depends_on.iter().any(|dependency_id| { + spec.steps.iter().any(|candidate_step| { + candidate_step.id == dependency_id.as_str() + && candidate_step.agent != review_step.agent + && !is_review_step(candidate_step) + }) + }) + }) + .collect::>(); + if paired_review_steps.is_empty() { return Err(WorkflowSpecError::invalid( - "draft workflows must include adversarial work by at least one agent or one step", + "draft workflows must include at least one independent adversarial review run assigned to a reviewer agent", )); } + + if !paired_review_steps + .iter() + .any(|step| verifies_finite_review_artifact(spec, step)) + { + return Err(WorkflowSpecError::invalid( + "single-review workflows must verifier-gate a finite review artifact contract containing candidate identity, acceptance criteria, verdict, blocking P0/P1 findings, non-blocking P2/P3 findings, and remediation cycle 0..2", + )); + } + Ok(()) } +fn is_review_step(step: &WorkflowStep) -> bool { + is_reviewer_text(&step.id) || is_adversarial_text(&step.title) +} + +fn verifies_finite_review_artifact(spec: &WorkflowSpec, step: &WorkflowStep) -> bool { + let Some(completion) = &step.completion else { + return false; + }; + completion.verifiers.iter().any(|verifier| { + if !verifier_has_finite_review_artifact_contract(verifier) { + return false; + } + let Some(artifact) = verifier.artifact.as_deref() else { + return false; + }; + step.outputs.iter().any(|output| output == artifact) + && spec + .artifacts + .required + .iter() + .any(|required| required == artifact) + }) +} + +/// Returns true when an `artifact_contains` verifier declares the finite review +/// artifact contract used by the single-review workflow path. +/// +/// This is exported so activation and orchestration enforce the same contract +/// the parser admitted instead of reimplementing a weaker string heuristic. +pub fn verifier_has_finite_review_artifact_contract(verifier: &WorkflowVerifier) -> bool { + verifier.kind == ARTIFACT_CONTAINS_VERIFIER + && REVIEW_ARTIFACT_CONTRACT_FIELDS.iter().all(|required| { + verifier + .must_contain + .iter() + .any(|field| field.trim() == *required) + }) +} + +fn is_reviewer_text(value: &str) -> bool { + is_adversarial_text(value) + || value + .split(|ch: char| !ch.is_ascii_alphanumeric()) + .any(|part| matches!(part.to_ascii_lowercase().as_str(), "review" | "reviewer")) +} + fn is_adversarial_text(value: &str) -> bool { let value = value.to_ascii_lowercase(); value.contains("adversary")