diff --git a/codex-rs/app-server/src/request_processors/background_agent_live.rs b/codex-rs/app-server/src/request_processors/background_agent_live.rs index ef924e177..5a0b80a6b 100644 --- a/codex-rs/app-server/src/request_processors/background_agent_live.rs +++ b/codex-rs/app-server/src/request_processors/background_agent_live.rs @@ -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> { + 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 { + 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 { + 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()), }; diff --git a/codex-rs/background-agent/src/lib.rs b/codex-rs/background-agent/src/lib.rs index 46aa79805..6494ffae9 100644 --- a/codex-rs/background-agent/src/lib.rs +++ b/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; diff --git a/codex-rs/background-agent/src/worker_admission.rs b/codex-rs/background-agent/src/worker_admission.rs new file mode 100644 index 000000000..e1b82bbf3 --- /dev/null +++ b/codex-rs/background-agent/src/worker_admission.rs @@ -0,0 +1,599 @@ +use crate::process_lifecycle::WorkerProcessCommand; +use anyhow::Context; +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value; +use std::ffi::OsString; +use std::future::Future; +use std::path::PathBuf; +use std::process::Stdio; +use tokio::process::Command; + +pub const CONVERSATIONS_AGENT_ID_ENV: &str = "CONVERSATIONS_AGENT_ID"; +pub const WORKER_ADMISSION_SNAPSHOT_FIELD: &str = "workerAdmission"; +const ROSTER_PAGE_SIZE: usize = 100; +const MAX_ROSTER_PAGES: usize = 1_000; + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct WorkerAdmissionInput { + pub worker: Option, + pub parent: Option, + pub task_id: Option, + pub artifact_type: Option, + pub artifact_id: Option, +} + +impl WorkerAdmissionInput { + pub fn into_request(self) -> anyhow::Result> { + if self == Self::default() { + return Ok(None); + } + Ok(Some(WorkerAdmissionRequest { + worker: required_field(self.worker, "worker")?, + parent: required_field(self.parent, "parent")?, + task_id: required_field(self.task_id, "task-id")?, + artifact_type: required_field(self.artifact_type, "artifact-type")?, + artifact_id: required_field(self.artifact_id, "artifact-id")?, + })) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerAdmissionRequest { + pub worker: String, + pub parent: String, + pub task_id: String, + pub artifact_type: String, + pub artifact_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkerAdmission { + pub worker: String, + pub parent: String, + pub task_id: String, + pub artifact_type: String, + pub artifact_id: String, + pub task_assignee: String, + pub worker_reports_to: String, + pub evidence: WorkerAdmissionEvidence, +} + +impl WorkerAdmission { + pub fn request(&self) -> WorkerAdmissionRequest { + WorkerAdmissionRequest { + worker: self.worker.clone(), + parent: self.parent.clone(), + task_id: self.task_id.clone(), + artifact_type: self.artifact_type.clone(), + artifact_id: self.artifact_id.clone(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkerAdmissionEvidence { + pub identities_worker_id: String, + pub todos_worker_id: String, + pub todos_parent_id: String, + pub conversations_worker_id: String, + pub effective_parent: String, + pub lock_holder: String, + pub roster_pages_scanned: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerAdmissionPrograms { + pub identities: PathBuf, + pub todos: PathBuf, + pub conversations: PathBuf, +} + +impl Default for WorkerAdmissionPrograms { + fn default() -> Self { + Self { + identities: PathBuf::from("identities"), + todos: PathBuf::from("todos"), + conversations: PathBuf::from("conversations"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerAdmissionCommand { + pub program: PathBuf, + pub args: Vec, + pub env: Vec<(OsString, OsString)>, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerAdmissionCommandOutput { + pub exit_code: i32, + pub stdout: Vec, + pub stderr: Vec, +} + +/// Runs one worker-admission dependency command without a shell. +/// +/// Implementations must preserve argv and environment field boundaries and +/// return stdout separately from stderr so JSON parsing never consumes a human +/// pagination footer. +pub trait WorkerAdmissionCommandRunner { + fn run( + &self, + command: WorkerAdmissionCommand, + ) -> impl Future> + Send; +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct ProcessWorkerAdmissionCommandRunner; + +impl WorkerAdmissionCommandRunner for ProcessWorkerAdmissionCommandRunner { + async fn run( + &self, + command: WorkerAdmissionCommand, + ) -> anyhow::Result { + let output = Command::new(&command.program) + .args(&command.args) + .envs(command.env) + .stdin(Stdio::null()) + .output() + .await + .with_context(|| { + format!( + "failed to execute worker-admission dependency {}", + command.program.display() + ) + })?; + Ok(WorkerAdmissionCommandOutput { + exit_code: output.status.code().unwrap_or(1), + stdout: output.stdout, + stderr: output.stderr, + }) + } +} + +pub async fn verify_worker_admission( + runner: &impl WorkerAdmissionCommandRunner, + programs: &WorkerAdmissionPrograms, + request: &WorkerAdmissionRequest, +) -> anyhow::Result { + verify_worker_admission_inner( + runner, + programs, + request, + EffectiveParentCheck::CurrentProcess, + ) + .await +} + +pub async fn revalidate_worker_admission( + runner: &impl WorkerAdmissionCommandRunner, + programs: &WorkerAdmissionPrograms, + admitted: &WorkerAdmission, +) -> anyhow::Result { + let current = verify_worker_admission_inner( + runner, + programs, + &admitted.request(), + EffectiveParentCheck::Persisted { + identity: admitted.evidence.effective_parent.as_str(), + }, + ) + .await?; + ensure_stable_evidence(admitted, ¤t)?; + Ok(current) +} + +pub fn worker_admission_from_snapshot(payload: &Value) -> anyhow::Result> { + payload + .get(WORKER_ADMISSION_SNAPSHOT_FIELD) + .filter(|value| !value.is_null()) + .cloned() + .map(serde_json::from_value) + .transpose() + .context("invalid persisted worker-admission evidence") +} + +pub fn apply_worker_identity( + command: WorkerProcessCommand, + admission: &WorkerAdmission, +) -> WorkerProcessCommand { + command.env(CONVERSATIONS_AGENT_ID_ENV, admission.worker.as_str()) +} + +enum EffectiveParentCheck<'a> { + CurrentProcess, + Persisted { identity: &'a str }, +} + +async fn verify_worker_admission_inner( + runner: &impl WorkerAdmissionCommandRunner, + programs: &WorkerAdmissionPrograms, + request: &WorkerAdmissionRequest, + effective_parent_check: EffectiveParentCheck<'_>, +) -> anyhow::Result { + validate_request(request)?; + let identities_worker = run_json( + runner, + WorkerAdmissionCommand { + program: programs.identities.clone(), + args: argv([ + "--json", + "show", + format!("agent:{}", request.worker).as_str(), + ]), + env: Vec::new(), + }, + &[0], + "read worker identity from Identities", + ) + .await + .with_context(|| { + format!( + "worker `{}` is not registered in Identities", + request.worker + ) + })?; + let identities_worker_id = identities_worker + .get("id") + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty()) + .context("worker Identities record has no id")? + .to_string(); + let identifier_matches = identities_worker + .pointer("/uniqueIdentifier/value") + .and_then(Value::as_str) + .is_some_and(|value| same_identity(value, request.worker.as_str())); + if !identifier_matches { + anyhow::bail!( + "worker Identities record does not resolve exact agent identity `{}`", + request.worker + ); + } + + let worker_agent = read_todos_agent(runner, programs, request.worker.as_str()).await?; + let parent_agent = read_todos_agent(runner, programs, request.parent.as_str()).await?; + let todos_worker_id = required_json_string(&worker_agent, "id", "worker Todos agent")?; + let todos_parent_id = required_json_string(&parent_agent, "id", "parent Todos agent")?; + let worker_name = required_json_string(&worker_agent, "name", "worker Todos agent")?; + let parent_name = required_json_string(&parent_agent, "name", "parent Todos agent")?; + if !same_identity(worker_name.as_str(), request.worker.as_str()) { + anyhow::bail!( + "Todos worker record resolved `{worker_name}` instead of `{}`", + request.worker + ); + } + if !same_identity(parent_name.as_str(), request.parent.as_str()) { + anyhow::bail!( + "Todos parent record resolved `{parent_name}` instead of `{}`", + request.parent + ); + } + let worker_reports_to = + required_json_string(&worker_agent, "reports_to", "worker Todos agent")?; + + let task = run_json( + runner, + WorkerAdmissionCommand { + program: programs.todos.clone(), + args: argv(["--json", "show", request.task_id.as_str()]), + env: Vec::new(), + }, + &[0], + "read declared Todos task", + ) + .await?; + let task_id = required_json_string(&task, "id", "Todos task")?; + if task_id != request.task_id { + anyhow::bail!( + "Todos task lookup returned `{task_id}` instead of `{}`", + request.task_id + ); + } + let task_status = required_json_string(&task, "status", "Todos task")?; + if !matches!(task_status.as_str(), "active" | "in_progress") { + anyhow::bail!( + "Todos task `{}` is not active/in_progress (status `{task_status}`)", + request.task_id + ); + } + let task_assignee = required_json_string(&task, "assigned_to", "Todos task")?; + if !same_identity(task_assignee.as_str(), request.parent.as_str()) { + anyhow::bail!( + "Todos task `{}` assigned_to `{task_assignee}` does not match parent `{}`", + request.task_id, + request.parent + ); + } + if worker_reports_to != todos_parent_id { + anyhow::bail!( + "worker `{}` reports_to `{worker_reports_to}` instead of parent Todos id `{todos_parent_id}`", + request.worker + ); + } + + let (conversations_worker_id, roster_pages_scanned) = + find_conversations_worker(runner, programs, request.worker.as_str()).await?; + let effective_parent = match effective_parent_check { + EffectiveParentCheck::CurrentProcess => { + let whoami = run_json( + runner, + WorkerAdmissionCommand { + program: programs.conversations.clone(), + args: argv(["whoami", "--json"]), + env: Vec::new(), + }, + &[0], + "read effective Conversations identity", + ) + .await?; + let effective_parent = required_json_string(&whoami, "agent", "Conversations whoami")?; + if !same_identity(effective_parent.as_str(), request.parent.as_str()) { + anyhow::bail!( + "effective Conversations identity `{effective_parent}` does not match parent `{}`", + request.parent + ); + } + effective_parent + } + EffectiveParentCheck::Persisted { identity } => identity.to_string(), + }; + + let lock = run_json( + runner, + WorkerAdmissionCommand { + program: programs.conversations.clone(), + args: argv([ + "locks", + "check", + request.artifact_id.as_str(), + "--type", + request.artifact_type.as_str(), + "--json", + ]), + env: Vec::new(), + }, + &[0, 2], + "check worker artifact lock", + ) + .await?; + let locked = lock + .get("locked") + .and_then(Value::as_bool) + .context("Conversations lock check returned no locked boolean")?; + if !locked { + anyhow::bail!( + "artifact {} `{}` is not actively locked", + request.artifact_type, + request.artifact_id + ); + } + let returned_artifact_type = + required_json_string(&lock, "resource_type", "Conversations lock")?; + let returned_artifact_id = required_json_string(&lock, "resource_id", "Conversations lock")?; + if returned_artifact_type != request.artifact_type + || returned_artifact_id != request.artifact_id + { + anyhow::bail!( + "Conversations lock evidence names {} `{}` instead of {} `{}`", + returned_artifact_type, + returned_artifact_id, + request.artifact_type, + request.artifact_id + ); + } + let lock_holder = required_json_string(&lock, "agent_id", "Conversations lock")?; + if !same_identity(lock_holder.as_str(), request.worker.as_str()) { + anyhow::bail!( + "artifact {} `{}` is held by `{lock_holder}`, not worker `{}`", + request.artifact_type, + request.artifact_id, + request.worker + ); + } + + Ok(WorkerAdmission { + worker: request.worker.clone(), + parent: request.parent.clone(), + task_id: request.task_id.clone(), + artifact_type: request.artifact_type.clone(), + artifact_id: request.artifact_id.clone(), + task_assignee, + worker_reports_to, + evidence: WorkerAdmissionEvidence { + identities_worker_id, + todos_worker_id, + todos_parent_id, + conversations_worker_id, + effective_parent, + lock_holder, + roster_pages_scanned, + }, + }) +} + +async fn read_todos_agent( + runner: &impl WorkerAdmissionCommandRunner, + programs: &WorkerAdmissionPrograms, + name: &str, +) -> anyhow::Result { + let response = run_json( + runner, + WorkerAdmissionCommand { + program: programs.todos.clone(), + args: argv(["--json", "agent", name]), + env: Vec::new(), + }, + &[0], + format!("read Todos agent `{name}`").as_str(), + ) + .await?; + response + .get("agent") + .filter(|value| value.is_object()) + .cloned() + .with_context(|| format!("Todos has no registered agent `{name}`")) +} + +async fn find_conversations_worker( + runner: &impl WorkerAdmissionCommandRunner, + programs: &WorkerAdmissionPrograms, + worker: &str, +) -> anyhow::Result<(String, usize)> { + let mut cursor = 0; + let mut pages_scanned = 0; + let mut matches = Vec::new(); + loop { + if pages_scanned >= MAX_ROSTER_PAGES { + anyhow::bail!("Conversations roster did not terminate after {MAX_ROSTER_PAGES} pages"); + } + let page = run_json( + runner, + WorkerAdmissionCommand { + program: programs.conversations.clone(), + args: argv([ + "agents", + "list", + "--json", + "--limit", + "100", + "--cursor", + cursor.to_string().as_str(), + ]), + env: Vec::new(), + }, + &[0], + format!("read Conversations roster page at cursor {cursor}").as_str(), + ) + .await + .with_context(|| format!("failed Conversations roster page at cursor {cursor}"))?; + let agents = page.as_array().with_context(|| { + format!("Conversations roster page at cursor {cursor} is not a JSON array") + })?; + pages_scanned += 1; + matches.extend(agents.iter().filter_map(|agent| { + let name = agent.get("agent").and_then(Value::as_str)?; + if same_identity(name, worker) { + agent.get("id").and_then(Value::as_str).map(str::to_string) + } else { + None + } + })); + if agents.len() < ROSTER_PAGE_SIZE { + break; + } + cursor += ROSTER_PAGE_SIZE; + } + match matches.as_slice() { + [] => anyhow::bail!( + "worker `{worker}` is not registered in the exhaustive Conversations roster" + ), + [id] => Ok((id.clone(), pages_scanned)), + _ => anyhow::bail!( + "worker `{worker}` has {} ambiguous Conversations registrations", + matches.len() + ), + } +} + +async fn run_json( + runner: &impl WorkerAdmissionCommandRunner, + command: WorkerAdmissionCommand, + accepted_exit_codes: &[i32], + action: &str, +) -> anyhow::Result { + let output = runner.run(command).await?; + if !accepted_exit_codes.contains(&output.exit_code) { + let stderr = String::from_utf8_lossy(&output.stderr); + anyhow::bail!( + "{action} failed with exit {}: {}", + output.exit_code, + stderr.trim() + ); + } + serde_json::from_slice(&output.stdout) + .with_context(|| format!("{action} returned invalid JSON on stdout")) +} + +fn ensure_stable_evidence( + admitted: &WorkerAdmission, + current: &WorkerAdmission, +) -> anyhow::Result<()> { + let admitted_stable = [ + admitted.worker.as_str(), + admitted.parent.as_str(), + admitted.task_id.as_str(), + admitted.artifact_type.as_str(), + admitted.artifact_id.as_str(), + admitted.task_assignee.as_str(), + admitted.worker_reports_to.as_str(), + admitted.evidence.identities_worker_id.as_str(), + admitted.evidence.todos_worker_id.as_str(), + admitted.evidence.todos_parent_id.as_str(), + admitted.evidence.conversations_worker_id.as_str(), + admitted.evidence.effective_parent.as_str(), + admitted.evidence.lock_holder.as_str(), + ]; + let current_stable = [ + current.worker.as_str(), + current.parent.as_str(), + current.task_id.as_str(), + current.artifact_type.as_str(), + current.artifact_id.as_str(), + current.task_assignee.as_str(), + current.worker_reports_to.as_str(), + current.evidence.identities_worker_id.as_str(), + current.evidence.todos_worker_id.as_str(), + current.evidence.todos_parent_id.as_str(), + current.evidence.conversations_worker_id.as_str(), + current.evidence.effective_parent.as_str(), + current.evidence.lock_holder.as_str(), + ]; + if admitted_stable != current_stable { + anyhow::bail!("worker-admission evidence changed before process spawn"); + } + Ok(()) +} + +fn validate_request(request: &WorkerAdmissionRequest) -> anyhow::Result<()> { + for (name, value) in [ + ("worker", request.worker.as_str()), + ("parent", request.parent.as_str()), + ("task-id", request.task_id.as_str()), + ("artifact-type", request.artifact_type.as_str()), + ("artifact-id", request.artifact_id.as_str()), + ] { + if value.trim().is_empty() { + anyhow::bail!("worker admission requires non-empty {name}"); + } + } + Ok(()) +} + +fn required_field(value: Option, name: &str) -> anyhow::Result { + let value = value.with_context(|| format!("worker admission requires --{name}"))?; + if value.trim().is_empty() { + anyhow::bail!("worker admission requires non-empty --{name}"); + } + Ok(value) +} + +fn required_json_string(value: &Value, field: &str, record: &str) -> anyhow::Result { + value + .get(field) + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty()) + .map(str::to_string) + .with_context(|| format!("{record} has no {field}")) +} + +fn same_identity(left: &str, right: &str) -> bool { + left.trim().eq_ignore_ascii_case(right.trim()) +} + +fn argv(values: [&str; N]) -> Vec { + values.into_iter().map(OsString::from).collect() +} diff --git a/codex-rs/background-agent/tests/worker_admission.rs b/codex-rs/background-agent/tests/worker_admission.rs new file mode 100644 index 000000000..34ae39c97 --- /dev/null +++ b/codex-rs/background-agent/tests/worker_admission.rs @@ -0,0 +1,633 @@ +use codex_background_agent::process_lifecycle::WorkerProcessCommand; +use codex_background_agent::worker_admission::CONVERSATIONS_AGENT_ID_ENV; +use codex_background_agent::worker_admission::WorkerAdmission; +use codex_background_agent::worker_admission::WorkerAdmissionCommand; +use codex_background_agent::worker_admission::WorkerAdmissionCommandOutput; +use codex_background_agent::worker_admission::WorkerAdmissionCommandRunner; +use codex_background_agent::worker_admission::WorkerAdmissionInput; +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::verify_worker_admission; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::collections::VecDeque; +use std::ffi::OsString; +use std::fmt::Debug; +use std::fmt::Display; +use std::future::Future; +use std::sync::Mutex; + +const WORKER: &str = "worker-one"; +const PARENT: &str = "parent-one"; +const TASK_ID: &str = "85553dae-7c24-4777-a281-b335f24b74ef"; +const ARTIFACT_TYPE: &str = "git-branch"; +const ARTIFACT_ID: &str = "github:hasna/codewith:branch:feature"; +const WORKER_TODOS_ID: &str = "worker-todos-id"; +const PARENT_TODOS_ID: &str = "parent-todos-id"; + +#[derive(Debug)] +struct ExpectedCommand { + program: &'static str, + args: Vec, + output: WorkerAdmissionCommandOutput, +} + +#[derive(Debug, Default)] +struct ScriptedRunner { + expected: Mutex>, +} + +impl ScriptedRunner { + fn new(expected: Vec) -> Self { + Self { + expected: Mutex::new(expected.into()), + } + } + + fn assert_exhausted(&self) { + assert_eq!( + lock_expected_commands(&self.expected).len(), + 0, + "all expected direct-argv commands should be consumed" + ); + } +} + +impl WorkerAdmissionCommandRunner for ScriptedRunner { + #[allow(clippy::manual_async_fn)] + fn run( + &self, + command: WorkerAdmissionCommand, + ) -> impl Future> + Send { + async move { + let expected = match lock_expected_commands(&self.expected).pop_front() { + Some(expected) => expected, + None => panic!("unexpected worker-admission command"), + }; + assert_eq!( + command.program.to_string_lossy(), + expected.program, + "program must be invoked directly" + ); + assert_eq!( + command + .args + .iter() + .map(|arg| arg.to_string_lossy().to_string()) + .collect::>(), + expected.args, + "argv must preserve field boundaries without shell interpolation" + ); + assert_eq!(command.env, Vec::<(OsString, OsString)>::new()); + Ok(expected.output) + } + } +} + +fn lock_expected_commands( + expected: &Mutex>, +) -> std::sync::MutexGuard<'_, VecDeque> { + match expected.lock() { + Ok(guard) => guard, + Err(error) => panic!("runner lock poisoned: {error}"), + } +} + +fn must_ok(result: Result, context: &str) -> T { + match result { + Ok(value) => value, + Err(error) => panic!("{context}: {error}"), + } +} + +fn must_err(result: Result, context: &str) -> E { + match result { + Ok(value) => panic!("{context}: unexpectedly succeeded with {value:?}"), + Err(error) => error, + } +} + +fn must_some(option: Option, context: &str) -> T { + match option { + Some(value) => value, + None => panic!("{context}"), + } +} + +fn output(exit_code: i32, stdout: Value) -> WorkerAdmissionCommandOutput { + WorkerAdmissionCommandOutput { + exit_code, + stdout: must_ok(serde_json::to_vec(&stdout), "serialize command output"), + stderr: Vec::new(), + } +} + +fn failed_output(stderr: &str) -> WorkerAdmissionCommandOutput { + WorkerAdmissionCommandOutput { + exit_code: 1, + stdout: Vec::new(), + stderr: stderr.as_bytes().to_vec(), + } +} + +fn input() -> WorkerAdmissionInput { + WorkerAdmissionInput { + worker: Some(WORKER.to_string()), + parent: Some(PARENT.to_string()), + task_id: Some(TASK_ID.to_string()), + artifact_type: Some(ARTIFACT_TYPE.to_string()), + artifact_id: Some(ARTIFACT_ID.to_string()), + } +} + +fn request() -> codex_background_agent::worker_admission::WorkerAdmissionRequest { + must_some( + must_ok(input().into_request(), "complete worker admission"), + "worker admission enabled", + ) +} + +fn identity_command(exit_code: i32) -> ExpectedCommand { + ExpectedCommand { + program: "identities", + args: vec![ + "--json".to_string(), + "show".to_string(), + format!("agent:{WORKER}"), + ], + output: if exit_code == 0 { + output( + /*exit_code*/ 0, + json!({ + "id": "worker-identities-id", + "fullName": WORKER, + "uniqueIdentifier": { + "scheme": "agent", + "value": WORKER + } + }), + ) + } else { + failed_output("identity not found") + }, + } +} + +fn todos_agent_command(name: &str, id: &str, reports_to: Option<&str>) -> ExpectedCommand { + ExpectedCommand { + program: "todos", + args: vec!["--json".to_string(), "agent".to_string(), name.to_string()], + output: output( + /*exit_code*/ 0, + json!({ + "agent": { + "id": id, + "name": name, + "status": "active", + "reports_to": reports_to + }, + "tasks": {}, + "all_tasks": [] + }), + ), + } +} + +fn task_command(assigned_to: &str, status: &str) -> ExpectedCommand { + ExpectedCommand { + program: "todos", + args: vec![ + "--json".to_string(), + "show".to_string(), + TASK_ID.to_string(), + ], + output: output( + /*exit_code*/ 0, + json!({ + "id": TASK_ID, + "status": status, + "assigned_to": assigned_to + }), + ), + } +} + +fn roster_command(cursor: usize, agents: Value) -> ExpectedCommand { + ExpectedCommand { + program: "conversations", + args: vec![ + "agents".to_string(), + "list".to_string(), + "--json".to_string(), + "--limit".to_string(), + "100".to_string(), + "--cursor".to_string(), + cursor.to_string(), + ], + output: output(/*exit_code*/ 0, agents), + } +} + +fn whoami_command(agent: &str) -> ExpectedCommand { + ExpectedCommand { + program: "conversations", + args: vec!["whoami".to_string(), "--json".to_string()], + output: output( + /*exit_code*/ 0, + json!({"agent": agent, "source": "env var"}), + ), + } +} + +fn lock_command( + resource_type: &str, + resource_id: &str, + locked: bool, + holder: &str, +) -> ExpectedCommand { + ExpectedCommand { + program: "conversations", + args: vec![ + "locks".to_string(), + "check".to_string(), + resource_id.to_string(), + "--type".to_string(), + resource_type.to_string(), + "--json".to_string(), + ], + output: output( + if locked { 2 } else { 0 }, + json!({ + "locked": locked, + "resource_type": resource_type, + "resource_id": resource_id, + "agent_id": holder, + "lock_type": "exclusive" + }), + ), + } +} + +fn valid_commands(roster_pages: Vec) -> Vec { + let mut commands = vec![ + identity_command(/*exit_code*/ 0), + todos_agent_command(WORKER, WORKER_TODOS_ID, Some(PARENT_TODOS_ID)), + todos_agent_command(PARENT, PARENT_TODOS_ID, /*reports_to*/ None), + task_command(PARENT, "in_progress"), + ]; + commands.extend( + roster_pages + .into_iter() + .enumerate() + .map(|(page, agents)| roster_command(page * 100, agents)), + ); + commands.extend([ + whoami_command(PARENT), + lock_command(ARTIFACT_TYPE, ARTIFACT_ID, /*locked*/ true, WORKER), + ]); + commands +} + +fn one_page_roster() -> Vec { + vec![json!([ + {"id": "parent-conversations-id", "agent": PARENT}, + {"id": "worker-conversations-id", "agent": WORKER} + ])] +} + +#[test] +fn worker_admission_is_disabled_only_when_every_field_is_omitted() { + assert_eq!( + must_ok( + WorkerAdmissionInput::default().into_request(), + "omitted worker admission" + ), + None + ); +} + +#[test] +fn worker_admission_refuses_each_missing_identity_or_artifact_field() { + let missing_fields = [ + ( + "worker", + WorkerAdmissionInput { + worker: None, + ..input() + }, + ), + ( + "parent", + WorkerAdmissionInput { + parent: None, + ..input() + }, + ), + ( + "task-id", + WorkerAdmissionInput { + task_id: None, + ..input() + }, + ), + ( + "artifact-type", + WorkerAdmissionInput { + artifact_type: None, + ..input() + }, + ), + ( + "artifact-id", + WorkerAdmissionInput { + artifact_id: None, + ..input() + }, + ), + ]; + for (field, candidate) in missing_fields { + let error = must_err( + candidate.into_request(), + "partial worker admission must fail closed", + ); + assert!( + error.to_string().contains(field), + "missing {field} error should name the field: {error:#}" + ); + } +} + +#[tokio::test] +async fn unregistered_worker_identity_is_rejected_before_other_queries() { + let runner = ScriptedRunner::new(vec![identity_command(/*exit_code*/ 1)]); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "unregistered identity must fail", + ); + + assert!(error.to_string().contains("not registered in Identities")); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn worker_without_parent_lineage_is_rejected() { + let runner = ScriptedRunner::new(vec![ + identity_command(/*exit_code*/ 0), + todos_agent_command(WORKER, WORKER_TODOS_ID, /*reports_to*/ None), + todos_agent_command(PARENT, PARENT_TODOS_ID, /*reports_to*/ None), + ]); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "missing lineage must fail", + ); + + assert!(error.to_string().contains("reports_to")); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn task_assigned_to_a_different_parent_is_rejected() { + let runner = ScriptedRunner::new(vec![ + identity_command(/*exit_code*/ 0), + todos_agent_command(WORKER, WORKER_TODOS_ID, Some(PARENT_TODOS_ID)), + todos_agent_command(PARENT, PARENT_TODOS_ID, /*reports_to*/ None), + task_command("another-parent", "in_progress"), + ]); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "task-parent mismatch must fail", + ); + + assert!(error.to_string().contains("assigned_to")); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn effective_conversations_parent_mismatch_is_rejected() { + let mut commands = valid_commands(one_page_roster()); + commands[5] = whoami_command("another-parent"); + let _lock_command = commands.pop(); + let runner = ScriptedRunner::new(commands); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "actual parent mismatch must fail", + ); + + assert!( + error + .to_string() + .contains("effective Conversations identity") + ); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn missing_artifact_lock_is_rejected() { + let mut commands = valid_commands(one_page_roster()); + commands[6] = lock_command(ARTIFACT_TYPE, ARTIFACT_ID, /*locked*/ false, ""); + let runner = ScriptedRunner::new(commands); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "missing artifact lock must fail", + ); + + assert!(error.to_string().contains("is not actively locked")); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn lock_in_a_different_resource_namespace_does_not_satisfy_admission() { + let mut commands = valid_commands(one_page_roster()); + commands[6] = lock_command(ARTIFACT_TYPE, ARTIFACT_ID, /*locked*/ false, ""); + let runner = ScriptedRunner::new(commands); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "wrong lock namespace must fail", + ); + + assert!(error.to_string().contains(ARTIFACT_TYPE)); + assert!(error.to_string().contains(ARTIFACT_ID)); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn worker_beyond_first_roster_page_is_registered() { + let first_page = (0..100) + .map(|index| json!({"id": format!("other-{index}"), "agent": format!("other-{index}")})) + .collect::>(); + let second_page = json!([ + {"id": "worker-conversations-id", "agent": WORKER} + ]); + let runner = ScriptedRunner::new(valid_commands(vec![json!(first_page), second_page])); + + let admission = must_ok( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "worker on the second page must pass", + ); + + assert_eq!(admission.evidence.roster_pages_scanned, 2); + assert_eq!( + admission.evidence.conversations_worker_id, + "worker-conversations-id" + ); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn roster_page_error_fails_closed() { + let first_page = (0..100) + .map(|index| json!({"id": format!("other-{index}"), "agent": format!("other-{index}")})) + .collect::>(); + let mut commands = vec![ + identity_command(/*exit_code*/ 0), + todos_agent_command(WORKER, WORKER_TODOS_ID, Some(PARENT_TODOS_ID)), + todos_agent_command(PARENT, PARENT_TODOS_ID, /*reports_to*/ None), + task_command(PARENT, "in_progress"), + roster_command(/*cursor*/ 0, json!(first_page)), + ]; + commands.push(ExpectedCommand { + program: "conversations", + args: vec![ + "agents".to_string(), + "list".to_string(), + "--json".to_string(), + "--limit".to_string(), + "100".to_string(), + "--cursor".to_string(), + "100".to_string(), + ], + output: failed_output("registry unavailable"), + }); + let runner = ScriptedRunner::new(commands); + + let error = must_err( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "roster page failure must fail closed", + ); + + assert!(error.to_string().contains("cursor 100")); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn complete_worker_admission_persists_structured_evidence() { + let runner = ScriptedRunner::new(valid_commands(one_page_roster())); + + let admission = must_ok( + verify_worker_admission(&runner, &WorkerAdmissionPrograms::default(), &request()).await, + "complete worker admission", + ); + + assert_eq!( + admission, + WorkerAdmission { + worker: WORKER.to_string(), + parent: PARENT.to_string(), + task_id: TASK_ID.to_string(), + artifact_type: ARTIFACT_TYPE.to_string(), + artifact_id: ARTIFACT_ID.to_string(), + task_assignee: PARENT.to_string(), + worker_reports_to: PARENT_TODOS_ID.to_string(), + evidence: codex_background_agent::worker_admission::WorkerAdmissionEvidence { + identities_worker_id: "worker-identities-id".to_string(), + todos_worker_id: WORKER_TODOS_ID.to_string(), + todos_parent_id: PARENT_TODOS_ID.to_string(), + conversations_worker_id: "worker-conversations-id".to_string(), + effective_parent: PARENT.to_string(), + lock_holder: WORKER.to_string(), + roster_pages_scanned: 1, + }, + } + ); + assert_eq!( + must_ok( + serde_json::from_value::(must_ok( + serde_json::to_value(&admission), + "serialize admission" + )), + "deserialize admission" + ), + admission + ); + runner.assert_exhausted(); +} + +#[tokio::test] +async fn pre_spawn_revalidation_detects_a_lock_lost_after_admission() { + let admission_runner = ScriptedRunner::new(valid_commands(one_page_roster())); + let admission = must_ok( + verify_worker_admission( + &admission_runner, + &WorkerAdmissionPrograms::default(), + &request(), + ) + .await, + "initial admission", + ); + admission_runner.assert_exhausted(); + + let mut revalidation_commands = valid_commands(one_page_roster()); + let _lock_command = revalidation_commands.pop(); + let _whoami_command = revalidation_commands.pop(); + revalidation_commands.push(lock_command( + ARTIFACT_TYPE, + ARTIFACT_ID, + /*locked*/ false, + "", + )); + let revalidation_runner = ScriptedRunner::new(revalidation_commands); + let error = must_err( + revalidate_worker_admission( + &revalidation_runner, + &WorkerAdmissionPrograms::default(), + &admission, + ) + .await, + "lost lock must prevent spawn", + ); + + assert!(error.to_string().contains("is not actively locked")); + revalidation_runner.assert_exhausted(); +} + +#[test] +fn worker_process_command_sets_explicit_conversations_identity() { + let admission = WorkerAdmission { + worker: WORKER.to_string(), + parent: PARENT.to_string(), + task_id: TASK_ID.to_string(), + artifact_type: ARTIFACT_TYPE.to_string(), + artifact_id: ARTIFACT_ID.to_string(), + task_assignee: PARENT.to_string(), + worker_reports_to: PARENT_TODOS_ID.to_string(), + evidence: codex_background_agent::worker_admission::WorkerAdmissionEvidence { + identities_worker_id: "worker-identities-id".to_string(), + todos_worker_id: WORKER_TODOS_ID.to_string(), + todos_parent_id: PARENT_TODOS_ID.to_string(), + conversations_worker_id: "worker-conversations-id".to_string(), + effective_parent: PARENT.to_string(), + lock_holder: WORKER.to_string(), + roster_pages_scanned: 1, + }, + }; + let command = apply_worker_identity( + WorkerProcessCommand::new("codewith", "worker.stderr.log"), + &admission, + ); + + assert_eq!( + command.env, + vec![( + OsString::from(CONVERSATIONS_AGENT_ID_ENV), + OsString::from(WORKER) + )] + ); +} diff --git a/codex-rs/cli/src/agent_cmd.rs b/codex-rs/cli/src/agent_cmd.rs index fa43960d0..428924d95 100644 --- a/codex-rs/cli/src/agent_cmd.rs +++ b/codex-rs/cli/src/agent_cmd.rs @@ -16,6 +16,10 @@ use codex_background_agent::daemon::BackgroundAgentDaemon; use codex_background_agent::daemon::BackgroundAgentDaemonPaths; use codex_background_agent::daemon::background_agent_daemon_state_dir; use codex_background_agent::daemon::ensure_supported_platform as ensure_background_agent_supported_platform; +use codex_background_agent::worker_admission::ProcessWorkerAdmissionCommandRunner; +use codex_background_agent::worker_admission::WorkerAdmissionPrograms; +use codex_background_agent::worker_admission::WorkerAdmissionRequest; +use codex_background_agent::worker_admission::verify_worker_admission; use codex_core::config::find_codex_home; use codex_protocol::models::PermissionProfile; use codex_state::BackgroundAgentExecutionSnapshotParams; @@ -76,6 +80,10 @@ pub(crate) enum AgentSubcommand { /// Enqueue a durable background-agent run. Start(AgentStartCommand), + /// Enqueue a durable external worker after identity, lineage, task, and lock admission. + #[command(name = "start-worker")] + StartWorker(AgentWorkerStartCommand), + /// List durable background-agent runs. List(AgentListCommand), @@ -122,6 +130,47 @@ pub(crate) struct AgentStartCommand { json: bool, } +#[derive(Debug, Args)] +pub(crate) struct AgentWorkerStartCommand { + #[command(flatten)] + start: AgentStartCommand, + + /// Registered identity the worker process will use. + #[arg(long = "worker")] + worker: String, + + /// Current effective Conversations identity dispatching the worker. + #[arg(long = "parent")] + parent: String, + + /// Active parent-owned Todos task that authorizes the worker. + #[arg(long = "task")] + task_id: String, + + /// Conversations lock resource type for the one canonical artifact. + #[arg(long = "artifact-type")] + artifact_type: String, + + /// Conversations lock resource id for the one canonical artifact. + #[arg(long = "artifact-id")] + artifact_id: String, +} + +impl AgentWorkerStartCommand { + fn into_parts(self) -> (AgentStartCommand, WorkerAdmissionRequest) { + ( + self.start, + WorkerAdmissionRequest { + worker: self.worker, + parent: self.parent, + task_id: self.task_id, + artifact_type: self.artifact_type, + artifact_id: self.artifact_id, + }, + ) + } +} + #[derive(Debug, Args)] pub(crate) struct AgentListCommand { /// Maximum number of runs to return. @@ -199,6 +248,27 @@ pub(crate) async fn run_agent_command( let output = start_agent( state_db.as_ref(), cmd, + /*worker_admission_request*/ None, + runtime_context.as_ref(), + auth_profile, + ) + .await?; + ( + output, + if json { + AgentPrintMode::Json + } else { + AgentPrintMode::Start + }, + ) + } + AgentSubcommand::StartWorker(cmd) => { + let (start, worker_admission_request) = cmd.into_parts(); + let json = start.json; + let output = start_agent( + state_db.as_ref(), + start, + Some(worker_admission_request), runtime_context.as_ref(), auth_profile, ) @@ -1003,6 +1073,7 @@ fn resolve_agent_start_auth_profile( async fn start_agent( state_db: &StateRuntime, cmd: AgentStartCommand, + worker_admission_request: Option, runtime_context: Option<&AgentStartRuntimeContext>, auth_profile: Option<&str>, ) -> anyhow::Result { @@ -1012,6 +1083,19 @@ async fn start_agent( anyhow::bail!("agent prompt must not be empty"); } + let worker_admission = match worker_admission_request { + Some(request) => Some( + verify_worker_admission( + &ProcessWorkerAdmissionCommandRunner, + &WorkerAdmissionPrograms::default(), + &request, + ) + .await + .context("durable worker admission rejected before run creation")?, + ), + None => None, + }; + ensure_background_agent_supported_platform()?; let daemon_output = background_agent_daemon()?.start().await?; @@ -1040,6 +1124,7 @@ async fn start_agent( "model": runtime_context.and_then(|context| context.model.as_deref()), "provider": runtime_context.and_then(|context| context.provider.as_deref()), "serviceTier": runtime_context.and_then(|context| context.service_tier.as_deref()), + "workerAdmission": &worker_admission, "recoveryPolicy": "abort_mid_turn_resume_at_safe_boundary", }); let config_fingerprint = StateRuntime::background_agent_identity_sha256( @@ -1050,6 +1135,7 @@ async fn start_agent( "prompt": prompt, "promptSha256": StateRuntime::background_agent_identity_sha256(prompt.as_bytes()), "promptSnapshotRef": prompt_snapshot_ref.as_str(), + "workerAdmission": &worker_admission, }); let snapshot_params = BackgroundAgentExecutionSnapshotParams { run_id: agent_id.clone(), @@ -1065,6 +1151,7 @@ async fn start_agent( "provider": runtime_context.and_then(|context| context.provider.as_deref()), "serviceTier": runtime_context .and_then(|context| context.service_tier.as_deref()), + "workerAdmission": &worker_admission, "authProfileIdentitySha256": auth_profile_ref.as_deref().map(|profile| { StateRuntime::background_agent_identity_sha256(profile.as_bytes()) }), @@ -1095,7 +1182,11 @@ async fn start_agent( parent_agent_run_id: None, spawn_linkage_json: None, auth_profile_ref: auth_profile_ref.clone(), - status_reason: Some("queued by codewith agent start".to_string()), + status_reason: Some(if worker_admission.is_some() { + "queued by codewith agent start-worker".to_string() + } else { + "queued by codewith agent start".to_string() + }), config_fingerprint: Some(config_fingerprint), version_fingerprint: Some(BACKGROUND_AGENT_ADMISSION_SCHEMA_VERSION.to_string()), };