fix(workflows): require one paired adversarial review

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::<String>()
+    );
+    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<String> {
+    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::<serde_json::Result<String>>()?;
+    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,29 +988,39 @@ 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
   max_tool_calls: 20
 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"
       model: "mock-model"
       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: []

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 }