feat: gate durable worker admission

codex-rs/app-server/src/request_processors/background_agent_live.rs

@@ -80,6 +80,12 @@ use codex_background_agent::process_lifecycle::WorkerProcessCommand;
 use codex_background_agent::process_lifecycle::WorkerProcessController;
 use codex_background_agent::process_lifecycle::WorkerProcessHandle;
 use codex_background_agent::process_lifecycle::WorkerProcessStatus;
+use codex_background_agent::worker_admission::ProcessWorkerAdmissionCommandRunner;
+use codex_background_agent::worker_admission::WorkerAdmission;
+use codex_background_agent::worker_admission::WorkerAdmissionPrograms;
+use codex_background_agent::worker_admission::apply_worker_identity;
+use codex_background_agent::worker_admission::revalidate_worker_admission;
+use codex_background_agent::worker_admission::worker_admission_from_snapshot;
 use codex_core::NewThread;
 use codex_core::StartThreadOptions;
 use codex_core::config::ConfigOverrides;
@@ -1920,7 +1926,7 @@ async fn reconcile_background_agents(
             .await?;
             continue;
         }
-        if !should_start_background_run(&run) {
+        if !should_start_in_process_background_run(&context.state_db, &run).await? {
             continue;
         }
         if !context
@@ -2063,6 +2069,24 @@ async fn reconcile_background_agent_worker_processes(
         else {
             continue;
         };
+        let worker_admission =
+            match revalidate_background_agent_worker_admission(&context, run.id.as_str()).await {
+                Ok(worker_admission) => worker_admission,
+                Err(err) => {
+                    fail_claimed_background_agent_worker_process(
+                        &context,
+                        run.id.as_str(),
+                        generation,
+                        "worker admission pre-spawn revalidation failed",
+                        &json!({
+                            "reason": "worker_admission_revalidation_failed",
+                            "error": err.to_string(),
+                        }),
+                    )
+                    .await?;
+                    continue;
+                }
+            };
         let stderr_log_path = background_agent_worker_stderr_log_path(&context, run.id.as_str());
         let command = WorkerProcessCommand::new(&context.codex_bin, &stderr_log_path)
             .arg(OsString::from("app-server"))
@@ -2078,6 +2102,10 @@ async fn reconcile_background_agent_worker_processes(
                 BACKGROUND_AGENT_WORKER_GENERATION_ENV,
                 generation.to_string(),
             );
+        let command = match worker_admission.as_ref() {
+            Some(admission) => apply_worker_identity(command, admission),
+            None => command,
+        };
         let handle = match WorkerProcessController::default().spawn(command).await {
             Ok(handle) => handle,
             Err(err) => {
@@ -2181,6 +2209,29 @@ async fn reconcile_background_agent_worker_processes(
     Ok(())
 }
 
+async fn revalidate_background_agent_worker_admission(
+    context: &BackgroundAgentProcessSupervisorContext,
+    run_id: &str,
+) -> anyhow::Result<Option<WorkerAdmission>> {
+    let snapshot = context
+        .state_db
+        .get_background_agent_initial_execution_snapshot(run_id)
+        .await?
+        .with_context(|| {
+            format!("background agent `{run_id}` is missing its initial execution context snapshot")
+        })?;
+    let Some(admission) = worker_admission_from_snapshot(&snapshot.payload_json)? else {
+        return Ok(None);
+    };
+    revalidate_worker_admission(
+        &ProcessWorkerAdmissionCommandRunner,
+        &WorkerAdmissionPrograms::default(),
+        &admission,
+    )
+    .await
+    .map(Some)
+}
+
 async fn fail_claimed_background_agent_worker_process(
     context: &BackgroundAgentProcessSupervisorContext,
     run_id: &str,
@@ -3122,6 +3173,29 @@ fn should_start_background_run(run: &BackgroundAgentRun) -> bool {
     true
 }
 
+async fn should_start_in_process_background_run(
+    state_db: &StateDbHandle,
+    run: &BackgroundAgentRun,
+) -> anyhow::Result<bool> {
+    if !should_start_background_run(run) {
+        return Ok(false);
+    }
+    Ok(!background_agent_run_has_worker_admission(state_db, run.id.as_str()).await?)
+}
+
+async fn background_agent_run_has_worker_admission(
+    state_db: &StateDbHandle,
+    run_id: &str,
+) -> anyhow::Result<bool> {
+    let Some(snapshot) = state_db
+        .get_background_agent_initial_execution_snapshot(run_id)
+        .await?
+    else {
+        return Ok(false);
+    };
+    Ok(worker_admission_from_snapshot(&snapshot.payload_json)?.is_some())
+}
+
 fn background_agent_worker_preclaimed_generation(
     run: &BackgroundAgentRun,
     supervisor_id: &str,
@@ -6278,6 +6352,42 @@ done
         Ok(())
     }
 
+    #[tokio::test]
+    async fn in_process_reconciler_refuses_worker_admission_runs() -> anyhow::Result<()> {
+        let temp = TempDir::new()?;
+        let state_db =
+            codex_state::StateRuntime::init(temp.path().to_path_buf(), "test-provider".to_string())
+                .await?;
+        seed_worker_admission_queued_run(state_db.as_ref(), /*run_id*/ "worker-admission-run")
+            .await?;
+
+        let run = state_db
+            .get_background_agent_run("worker-admission-run")
+            .await?
+            .expect("seeded run should exist");
+        assert!(
+            should_start_background_run(&run),
+            "the old in-process reconciler predicate treated the queued run as claimable"
+        );
+        assert!(
+            state_db
+                .background_agent_admission_is_ready(
+                    /*run_id*/ "worker-admission-run",
+                    BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION,
+                    BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT,
+                )
+                .await?,
+            "the bypass must not rely on runtime compatibility failure"
+        );
+
+        assert!(
+            !should_start_in_process_background_run(&state_db, &run).await?,
+            "worker-admission runs require the process supervisor's revalidation and identity injection"
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn initial_goal_objective_payload_parser_trims_and_ignores_missing_values() {
         assert_eq!(
@@ -6974,6 +7084,61 @@ done
     async fn seed_queued_run(
         state_db: &codex_state::StateRuntime,
         run_id: &str,
+    ) -> anyhow::Result<()> {
+        seed_queued_run_with_payload(
+            state_db,
+            run_id,
+            json!({
+                "cwd": null,
+                "configFingerprint": "cfg-test",
+                "versionFingerprint": BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION,
+                "packageFingerprint": BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT,
+                "recoveryPolicy": "abort_mid_turn_resume_at_safe_boundary",
+            }),
+        )
+        .await
+    }
+
+    async fn seed_worker_admission_queued_run(
+        state_db: &codex_state::StateRuntime,
+        run_id: &str,
+    ) -> anyhow::Result<()> {
+        seed_queued_run_with_payload(
+            state_db,
+            run_id,
+            json!({
+                "cwd": null,
+                "configFingerprint": "cfg-test",
+                "versionFingerprint": BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION,
+                "packageFingerprint": BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT,
+                "recoveryPolicy": "abort_mid_turn_resume_at_safe_boundary",
+                "workerAdmission": {
+                    "worker": "worker-one",
+                    "parent": "parent-one",
+                    "taskId": "task-one",
+                    "artifactType": "git-branch",
+                    "artifactId": "github:hasna/codewith:branch:feature",
+                    "taskAssignee": "parent-one",
+                    "workerReportsTo": "parent-todos-id",
+                    "evidence": {
+                        "identitiesWorkerId": "worker-identities-id",
+                        "todosWorkerId": "worker-todos-id",
+                        "todosParentId": "parent-todos-id",
+                        "conversationsWorkerId": "worker-conversations-id",
+                        "effectiveParent": "parent-one",
+                        "lockHolder": "worker-one",
+                        "rosterPagesScanned": 1
+                    }
+                }
+            }),
+        )
+        .await
+    }
+
+    async fn seed_queued_run_with_payload(
+        state_db: &codex_state::StateRuntime,
+        run_id: &str,
+        execution_payload_json: Value,
     ) -> anyhow::Result<()> {
         let start_event_payload = json!({
             "cwd": null,
@@ -6983,13 +7148,7 @@ done
         let execution_snapshot_params = BackgroundAgentExecutionSnapshotParams {
             run_id: run_id.to_string(),
             snapshot_kind: "initial_execution_context".to_string(),
-            payload_json: json!({
-                "cwd": null,
-                "configFingerprint": "cfg-test",
-                "versionFingerprint": BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION,
-                "packageFingerprint": BACKGROUND_AGENT_RUNTIME_COMPATIBILITY_FINGERPRINT,
-                "recoveryPolicy": "abort_mid_turn_resume_at_safe_boundary",
-            }),
+            payload_json: execution_payload_json,
             recovery_policy: "abort_mid_turn_resume_at_safe_boundary".to_string(),
             config_fingerprint: Some("cfg-test".to_string()),
         };

codex-rs/background-agent/src/lib.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 pub mod daemon;
 pub mod process_lifecycle;
 mod supervisor;
+pub mod worker_admission;
 
 pub use codex_state::BACKGROUND_AGENT_EVENT_CURSOR_COMPACTED;
 pub use codex_state::BackgroundAgentDesiredState;