diff --git a/crates/homeboy-agents/src/agent_task_lifecycle/tests/submit_and_persist.rs b/crates/homeboy-agents/src/agent_task_lifecycle/tests/submit_and_persist.rs index 6ca41ced9..73fa0144a 100644 --- a/crates/homeboy-agents/src/agent_task_lifecycle/tests/submit_and_persist.rs +++ b/crates/homeboy-agents/src/agent_task_lifecycle/tests/submit_and_persist.rs @@ -1100,6 +1100,13 @@ fn terminal_daemon_status_waits_for_delayed_aggregate_then_projects_once() { .expect("terminal transport awaits aggregate synchronization"); assert_eq!(record.state, AgentTaskRunState::Running); assert!(store::read_aggregate(&record.run_id).is_err()); + let before_delayed_provider_terminal = status(&record.run_id) + .expect("continuation reconciliation observes the pending provider aggregate"); + assert_eq!( + before_delayed_provider_terminal.state, + AgentTaskRunState::Running, + "continuation must keep observing until the runner publishes the provider aggregate" + ); let aggregate = succeeded_aggregate(&test_plan()); let mut terminal_with_aggregate = terminal_child_snapshot(&aggregate); @@ -1123,6 +1130,13 @@ fn terminal_daemon_status_waits_for_delayed_aggregate_then_projects_once() { store::read_aggregate(&record.run_id).expect("aggregate persisted"), aggregate ); + let after_delayed_provider_terminal = status(&record.run_id) + .expect("continuation reconciliation observes the projected terminal aggregate"); + assert_eq!( + after_delayed_provider_terminal.state, + AgentTaskRunState::Succeeded, + "the same durable attempt becomes terminal exactly once when delayed provider evidence arrives" + ); }); } diff --git a/crates/homeboy-agents/src/agent_task_service/cook_recipe.rs b/crates/homeboy-agents/src/agent_task_service/cook_recipe.rs index 719dcc680..7b5b5c0ee 100644 --- a/crates/homeboy-agents/src/agent_task_service/cook_recipe.rs +++ b/crates/homeboy-agents/src/agent_task_service/cook_recipe.rs @@ -72,6 +72,15 @@ pub struct ClaimedCookContinuation { path: PathBuf, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CookContinuationState { + Absent, + Pending, + Claimed, + Failed, + Completed, +} + impl ClaimedCookContinuation { pub fn continuation(&self) -> &AgentTaskCookContinuation { &self.continuation @@ -1057,9 +1066,39 @@ pub fn claim_continuation_for( })) } -/// Atomically claim one exact continuation for explicit operator recovery. -/// Unlike the generic queue path, this never exposes the selected work as -/// pending for a background consumer to steal between rearm and claim. +/// Read the durable state of one exact continuation after reclaiming abandoned +/// claims. Callers use this to distinguish work owned by another consumer from +/// completed work and an explicitly recoverable failure. +pub fn continuation_state(cook_id: &str, run_id: &str) -> Result { + let hash = content_hash::sha256_hex(format!("{cook_id}:{run_id}").as_bytes()); + let root = queue_root()?; + fs::create_dir_all(&root) + .map_err(|error| Error::internal_io(error.to_string(), Some(root.display().to_string())))?; + reclaim_dead_claims(&root)?; + let claimed_prefix = format!("{hash}.claimed."); + if fs::read_dir(&root) + .map_err(|error| Error::internal_io(error.to_string(), Some(root.display().to_string())))? + .filter_map(std::result::Result::ok) + .filter_map(|entry| entry.file_name().into_string().ok()) + .any(|name| name.starts_with(&claimed_prefix)) + { + return Ok(CookContinuationState::Claimed); + } + for (suffix, state) in [ + ("pending", CookContinuationState::Pending), + ("failed", CookContinuationState::Failed), + ("completed", CookContinuationState::Completed), + ] { + if root.join(format!("{hash}.{suffix}")).exists() { + return Ok(state); + } + } + Ok(CookContinuationState::Absent) +} + +/// Atomically claim an existing failed continuation for explicit operator +/// recovery. Unlike the generic queue path, this never exposes the selected +/// work as pending for a background consumer to steal between rearm and claim. pub fn claim_continuation_for_recovery( cook_id: &str, run_id: &str, @@ -1077,14 +1116,8 @@ pub fn claim_continuation_for_recovery( None, )); } - let continuation = AgentTaskCookContinuation { - schema: CONTINUATION_SCHEMA.to_string(), - key: format!("{cook_id}:{run_id}"), - cook_id: cook_id.to_string(), - run_id: run_id.to_string(), - retries: 0, - }; - let hash = content_hash::sha256_hex(continuation.key.as_bytes()); + let key = format!("{cook_id}:{run_id}"); + let hash = content_hash::sha256_hex(key.as_bytes()); let root = queue_root()?; fs::create_dir_all(&root) .map_err(|error| Error::internal_io(error.to_string(), Some(root.display().to_string())))?; @@ -1095,93 +1128,103 @@ pub fn claim_continuation_for_recovery( .filter_map(|entry| entry.file_name().into_string().ok()) .any(|name| name.starts_with(&format!("{hash}.claimed."))) { - return Ok(None); + return Err(rearm_state_error( + cook_id, + run_id, + CookContinuationState::Claimed, + )); } let claimed = root.join(format!("{hash}.claimed.{}", std::process::id())); - for state in ["pending", "failed"] { - let source = root.join(format!("{hash}.{state}")); - match fs::rename(&source, &claimed) { - Ok(()) => { - fs::write( - &claimed, - serde_json::to_vec(&continuation) - .map_err(|error| Error::internal_json(error.to_string(), None))?, - ) - .map_err(|error| { - Error::internal_io(error.to_string(), Some(claimed.display().to_string())) - })?; - return Ok(Some(ClaimedCookContinuation { - continuation, - path: claimed, - })); - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => { - return Err(Error::internal_io( - error.to_string(), - Some(source.display().to_string()), - )) - } - } - } - - let completed = root.join(format!("{hash}.completed")); - if completed.exists() { - let record = agent_task_lifecycle::status(run_id)?; - if record - .metadata - .pointer("/latest_promotion/status") - .and_then(Value::as_str) - != Some("verification_pending") - { - return Ok(None); + let failed = root.join(format!("{hash}.failed")); + match fs::rename(&failed, &claimed) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Err(rearm_state_error( + cook_id, + run_id, + continuation_state(cook_id, run_id)?, + )) } - fs::rename(&completed, &claimed).map_err(|error| { - Error::internal_io(error.to_string(), Some(completed.display().to_string())) - })?; - fs::write( - &claimed, - serde_json::to_vec(&continuation) - .map_err(|error| Error::internal_json(error.to_string(), None))?, - ) - .map_err(|error| { - Error::internal_io(error.to_string(), Some(claimed.display().to_string())) - })?; - return Ok(Some(ClaimedCookContinuation { - continuation, - path: claimed, - })); - } - - let mut file = match OpenOptions::new() - .write(true) - .create_new(true) - .open(&claimed) - { - Ok(file) => file, - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => return Ok(None), Err(error) => { return Err(Error::internal_io( error.to_string(), - Some(claimed.display().to_string()), + Some(failed.display().to_string()), )) } - }; - file.write_all( - &serde_json::to_vec(&continuation) + } + let mut continuation: AgentTaskCookContinuation = + match read_claimed_continuation(&claimed, "failed durable continuation") { + Ok(continuation) => continuation, + Err(error) => { + fail_claimed_path(&claimed, &error.message)?; + return Err(error); + } + }; + if let Err(error) = validate_continuation(&continuation) { + fail_claimed_path(&claimed, &error.message)?; + return Err(error); + } + if continuation.key != key { + let error = Error::validation_invalid_argument( + "cook_continuation", + "failed durable continuation key does not match its Cook attempt", + Some(continuation.key.clone()), + None, + ); + fail_claimed_path(&claimed, &error.message)?; + return Err(error); + } + continuation.retries = 0; + fs::write( + &claimed, + serde_json::to_vec(&continuation) .map_err(|error| Error::internal_json(error.to_string(), None))?, ) .map_err(|error| Error::internal_io(error.to_string(), Some(claimed.display().to_string())))?; - file.sync_all().map_err(|error| { - Error::internal_io(error.to_string(), Some(claimed.display().to_string())) - })?; Ok(Some(ClaimedCookContinuation { continuation, path: claimed, })) } +fn rearm_state_error(cook_id: &str, run_id: &str, state: CookContinuationState) -> Error { + Error::validation_invalid_argument( + "cook_continuation.rearm", + format!( + "Cook `{cook_id}` attempt `{run_id}` cannot be rearmed because its continuation is {state:?}; only an existing failed continuation can be rearmed" + ), + Some(run_id.to_string()), + None, + ) +} + +fn read_claimed_continuation( + path: &std::path::Path, + description: &str, +) -> Result { + let metadata = fs::metadata(path) + .map_err(|error| Error::internal_io(error.to_string(), Some(path.display().to_string())))?; + if !metadata.is_file() { + return Err(Error::validation_invalid_argument( + "cook_continuation", + format!("unreadable {description}: expected a regular file"), + Some(path.display().to_string()), + None, + )); + } + let raw = fs::read(path) + .map_err(|error| Error::internal_io(error.to_string(), Some(path.display().to_string())))?; + serde_json::from_slice(&raw).map_err(|error| { + Error::validation_invalid_argument( + "cook_continuation", + format!("malformed {description}: {error}"), + Some(path.display().to_string()), + None, + ) + }) +} + pub fn reconstruct_options(recipe: &AgentTaskCookRecipe) -> Result { reconstruct_options_with_dispatcher(recipe, None) } @@ -1806,6 +1849,18 @@ fn reclaim_dead_claims(root: &std::path::Path) -> Result<()> { continue; }; if !homeboy_core::process::pid_is_running(pid) { + let continuation: AgentTaskCookContinuation = + match read_claimed_continuation(&path, "durable continuation") { + Ok(continuation) => continuation, + Err(error) => { + fail_claimed_path(&path, &error.message)?; + continue; + } + }; + if let Err(error) = validate_continuation(&continuation) { + fail_claimed_path(&path, &error.message)?; + continue; + } fs::rename(&path, continuation_state_path(&path, "pending")).map_err(|error| { Error::internal_io(error.to_string(), Some(path.display().to_string())) })?; @@ -2338,55 +2393,79 @@ mod tests { .fail("wrong controller runtime") .unwrap(); + assert_eq!( + continuation_state("cook", "run").unwrap(), + CookContinuationState::Failed + ); assert!(!enqueue_terminal_continuation("cook", "run").unwrap()); assert!(rearm_failed_terminal_continuation("cook", "run").unwrap()); + assert_eq!( + continuation_state("cook", "run").unwrap(), + CookContinuationState::Pending + ); let claim = claim_continuation_for("cook", "run") .unwrap() .expect("explicit recovery rearmed the failed continuation"); + assert_eq!( + continuation_state("cook", "run").unwrap(), + CookContinuationState::Claimed + ); assert_eq!(claim.continuation().retries, 0); claim.complete().unwrap(); + assert_eq!( + continuation_state("cook", "run").unwrap(), + CookContinuationState::Completed + ); assert!(!rearm_failed_terminal_continuation("cook", "run").unwrap()); }); } #[test] - fn targeted_recovery_claim_never_exposes_fresh_work_to_generic_consumers() { + fn targeted_recovery_claim_keeps_failed_work_owned_during_a_daemon_race() { homeboy_core::test_support::with_isolated_home(|_| { write_recipe(&recipe()).unwrap(); + enqueue_terminal_continuation("cook", "run").unwrap(); + claim_continuation_for("cook", "run") + .unwrap() + .unwrap() + .fail("promotion failed") + .unwrap(); let claim = claim_continuation_for_recovery("cook", "run") .unwrap() - .expect("operator owns exact continuation"); + .expect("operator atomically owns the failed continuation"); assert_eq!(claim.continuation().key, "cook:run"); - assert!(claim_continuation().unwrap().is_none()); + assert!( + claim_continuation().unwrap().is_none(), + "the daemon cannot steal a failed continuation while explicit recovery owns it" + ); claim.complete().unwrap(); }); } #[test] - fn targeted_recovery_reclaims_completed_but_still_unverified_work() { + fn targeted_recovery_rejects_absent_pending_and_completed_continuations() { homeboy_core::test_support::with_isolated_home(|_| { - persist_recipe_run(); + write_recipe(&recipe()).unwrap(); + let absent = claim_continuation_for_recovery("cook", "run") + .expect_err("absent work cannot be rearmed"); + assert!(absent.message.contains("Absent")); + enqueue_terminal_continuation("cook", "run").unwrap(); + let pending = claim_continuation_for_recovery("cook", "run") + .expect_err("pending work cannot be rearmed"); + assert!(pending.message.contains("Pending")); + claim_continuation_for("cook", "run") .unwrap() .unwrap() .complete() .unwrap(); - crate::agent_task_lifecycle::rewrite_record_for_test("run", |record| { - record.metadata["latest_promotion"] = - serde_json::json!({ "status": "verification_pending" }); - }) - .unwrap(); - - let claim = claim_continuation_for_recovery("cook", "run") - .unwrap() - .expect("unverified completed claim is recoverable"); - - assert_eq!(claim.continuation().key, "cook:run"); - claim.complete().unwrap(); + let completed = claim_continuation_for_recovery("cook", "run") + .expect_err("completed work cannot be rearmed"); + assert!(completed.message.contains("Completed")); }); } @@ -2473,6 +2552,66 @@ mod tests { }); } + #[test] + fn malformed_failed_recovery_is_terminalized_while_its_claim_is_live() { + homeboy_core::test_support::with_isolated_home(|_| { + write_recipe(&recipe()).unwrap(); + let root = queue_root().unwrap(); + fs::create_dir_all(&root).unwrap(); + let hash = content_hash::sha256_hex(b"cook:run"); + fs::write(root.join(format!("{hash}.failed")), b"not json").unwrap(); + + let error = claim_continuation_for_recovery("cook", "run") + .expect_err("malformed failed work must be terminalized after its atomic claim"); + + assert!(error + .message + .contains("malformed failed durable continuation")); + assert!(root.join(format!("{hash}.failed")).exists()); + assert!(fs::read_to_string(root.join(format!("{hash}.diagnostic"))) + .unwrap() + .contains("malformed failed durable continuation")); + assert!(!root.join(format!("{hash}.pending")).exists()); + assert!(!root + .join(format!("{hash}.claimed.{}", std::process::id())) + .exists()); + }); + } + + #[test] + fn unreadable_stale_recovery_claim_is_terminalized_without_requeueing() { + homeboy_core::test_support::with_isolated_home(|_| { + write_recipe(&recipe()).unwrap(); + let root = queue_root().unwrap(); + fs::create_dir_all(&root).unwrap(); + let hash = content_hash::sha256_hex(b"cook:run"); + // A directory is unreadable as a continuation payload on every + // supported platform while still allowing the stale claim rename. + fs::create_dir(root.join(format!("{hash}.claimed.4294967295"))).unwrap(); + + let error = claim_continuation_for_recovery("cook", "run") + .expect_err("stale unreadable work must remain terminal failed"); + + assert!( + error + .message + .contains("unreadable failed durable continuation"), + "{error:?}" + ); + assert!(root.join(format!("{hash}.failed")).exists()); + assert!(root.join(format!("{hash}.diagnostic")).exists()); + assert!(!root.join(format!("{hash}.pending")).exists()); + assert!( + !fs::read_dir(&root) + .unwrap() + .filter_map(std::result::Result::ok) + .filter_map(|entry| entry.file_name().into_string().ok()) + .any(|name| name.starts_with(&format!("{hash}.claimed."))), + "a stale malformed claim must not remain claimed" + ); + }); + } + #[test] fn failed_and_cancelled_attempts_do_not_enqueue_continuations() { homeboy_core::test_support::with_isolated_home(|_| { diff --git a/crates/homeboy-agents/src/agent_tasks.rs b/crates/homeboy-agents/src/agent_tasks.rs index 9e1c8acf8..f81ee187b 100644 --- a/crates/homeboy-agents/src/agent_tasks.rs +++ b/crates/homeboy-agents/src/agent_tasks.rs @@ -461,7 +461,7 @@ pub mod service { artifacts, authorize_cook_continue_route, cancel, claim_continuation_for, claim_continuation_for_recovery, compile_cook_attempt, compile_cook_attempt_with_readiness_cache, consume_claimed_terminal_with_dispatcher, - consume_claimed_with_dispatcher, cook_batch_job_submission, + consume_claimed_with_dispatcher, continuation_state, cook_batch_job_submission, detached_batch_coordinator_control, discover_runs, enqueue_terminal_continuation, evidence_ref_task_id, execute_promotion, hydrate_evidence_ref, hydrate_evidence_summary, load_recipe, load_recipe_for_attempt, logs, normalize_plan_workspaces, @@ -488,9 +488,10 @@ pub mod service { AgentTaskDiscoveryFilter, AgentTaskDiscoveryReport, AgentTaskDiscoveryRun, AgentTaskHydratedEvidence, AgentTaskPromotionJob, AgentTaskPromotionJobDriver, AgentTaskPromotionJobPhase, AgentTaskPromotionRequest, AgentTaskRetryServiceResult, - AgentTaskRunResult, CookActivityProbe, CookBatchJobDriver, CookProgressEvent, - CookProviderActivity, CookSupervisionTick, CookSupervisor, AGENT_TASK_COOK_BATCH_JOB_TYPE, - AGENT_TASK_COOK_BATCH_JOB_VERSION, AGENT_TASK_PROMOTION_JOB_TYPE, - AGENT_TASK_PROMOTION_JOB_VERSION, DETACHED_BATCH_COORDINATOR_ENV, + AgentTaskRunResult, CookActivityProbe, CookBatchJobDriver, CookContinuationState, + CookProgressEvent, CookProviderActivity, CookSupervisionTick, CookSupervisor, + AGENT_TASK_COOK_BATCH_JOB_TYPE, AGENT_TASK_COOK_BATCH_JOB_VERSION, + AGENT_TASK_PROMOTION_JOB_TYPE, AGENT_TASK_PROMOTION_JOB_VERSION, + DETACHED_BATCH_COORDINATOR_ENV, }; } diff --git a/crates/homeboy-cli/src/commands/agent_task/args/definitions/command.rs b/crates/homeboy-cli/src/commands/agent_task/args/definitions/command.rs index c6b625bb0..20349737b 100644 --- a/crates/homeboy-cli/src/commands/agent_task/args/definitions/command.rs +++ b/crates/homeboy-cli/src/commands/agent_task/args/definitions/command.rs @@ -222,6 +222,9 @@ pub struct CookContinueArgs { /// Validate continuation admission without dispatching a provider or mutating lifecycle state. #[arg(long)] pub preflight: bool, + /// Explicitly rearm one failed terminal continuation before consuming it. + #[arg(long, conflicts_with = "preflight")] + pub rearm: bool, /// Include the complete Cook report rather than the compact lifecycle view. #[arg(long)] pub full: bool, diff --git a/crates/homeboy-cli/src/commands/agent_task/run.rs b/crates/homeboy-cli/src/commands/agent_task/run.rs index 7fe8be796..0b4690453 100644 --- a/crates/homeboy-cli/src/commands/agent_task/run.rs +++ b/crates/homeboy-cli/src/commands/agent_task/run.rs @@ -139,6 +139,26 @@ pub(crate) fn run_cook(args: AgentTaskCookArgs) -> CmdResult { /// Resume a Cook from its immutable recipe rather than asking the operator to /// replay prompt, provider, gate, workspace, or disclosure arguments. pub(crate) fn continue_cook(args: CookContinueArgs) -> CmdResult { + continue_cook_with( + args, + ExtensionProviderAgentTaskExecutor::discover(), + crate::commands::infra::route::reconstruct_cook_attempt_dispatcher, + ) +} + +pub(crate) fn continue_cook_with( + args: CookContinueArgs, + executor: E, + reconstruct_dispatcher: F, +) -> CmdResult +where + E: AgentTaskExecutorAdapter + Clone, + F: Fn( + &Value, + ) -> homeboy::core::Result< + Option>, + > + Copy, +{ let recipe = agent_task_service::load_recipe(&args.cook_or_attempt_id).or_else(|cook_error| { agent_task_service::load_recipe_for_attempt(&args.cook_or_attempt_id)?.ok_or(cook_error) @@ -146,10 +166,7 @@ pub(crate) fn continue_cook(args: CookContinueArgs) -> CmdResult { let run_id = agent_task_service::resolve_cook_continuation_run_id(&args.cook_or_attempt_id)?; let record = agent_task_service::reconcile_recipe_attempt_for_continuation(&recipe, &run_id)?; if !record.state.is_terminal() { - return Ok(( - cook_continuation_status(&recipe.cook_id, &run_id, &format!("{:?}", record.state)), - 0, - )); + return Ok((cook_continuation_status(&recipe.cook_id, &record), 0)); } if matches!( @@ -158,29 +175,36 @@ pub(crate) fn continue_cook(args: CookContinueArgs) -> CmdResult { | agent_task_lifecycle::AgentTaskRunState::CandidateRecoverable | agent_task_lifecycle::AgentTaskRunState::PartialRecoverable ) { - // Explicit recovery claims the exact Cook without exposing it to the - // generic daemon queue between rearm and execution. - let Some(claim) = + // Ordinary continuation consumes only the pending entry scheduled by + // authoritative reconciliation. Failed and completed claims require an + // explicit rearm and can never be silently replayed. + let claim = if args.rearm { agent_task_service::claim_continuation_for_recovery(&recipe.cook_id, &run_id)? - else { + } else { + agent_task_service::claim_continuation_for(&recipe.cook_id, &run_id)? + }; + let Some(claim) = claim else { return Ok(( - cook_continuation_pending(&recipe.cook_id, &run_id, &format!("{:?}", record.state)), + cook_terminal_continuation_status( + &recipe.cook_id, + &run_id, + &format!("{:?}", record.state), + agent_task_service::continuation_state(&recipe.cook_id, &run_id)?, + ), 0, )); }; let mut result = None; let historical_terminal = recipe.runtime_generation != homeboy::core::build_identity::current().display; - let dispatcher = |dispatch_recipe: &Value| { - crate::commands::infra::route::reconstruct_cook_attempt_dispatcher(dispatch_recipe) - }; + let dispatcher = reconstruct_dispatcher; + let executor = executor.clone(); let execute = |options| { agent_task_service::authorize_cook_continue_route(&options)?; - let executor = ExtensionProviderAgentTaskExecutor::discover(); let cook = if historical_terminal { - agent_task_service::run_terminal_cook_continuation(options, executor)? + agent_task_service::run_terminal_cook_continuation(options, executor.clone())? } else { - agent_task_service::run_cook(options, executor)? + agent_task_service::run_cook(options, executor.clone())? }; let exit_code = cook.exit_code; result = Some(cook.value); @@ -207,9 +231,7 @@ pub(crate) fn continue_cook(args: CookContinueArgs) -> CmdResult { )); } - let dispatcher = crate::commands::infra::route::reconstruct_cook_attempt_dispatcher( - &recipe.promotion_transport["attempt_dispatch"], - )?; + let dispatcher = reconstruct_dispatcher(&recipe.promotion_transport["attempt_dispatch"])?; let attempt = recipe .attempts .iter() @@ -232,7 +254,6 @@ pub(crate) fn continue_cook(args: CookContinueArgs) -> CmdResult { options.initial_run_id = attempt.run_id.clone(); options.initial_plan = attempt.plan.clone(); agent_task_service::authorize_cook_continue_route(&options)?; - let executor = ExtensionProviderAgentTaskExecutor::discover(); let result = if terminal_review_form_continuation { agent_task_service::run_terminal_cook_continuation(options, executor)? } else { @@ -431,27 +452,126 @@ fn cook_continuation_preflight_report( }) } -fn cook_continuation_pending(cook_id: &str, run_id: &str, provider_state: &str) -> Value { +fn cook_terminal_continuation_status( + cook_id: &str, + run_id: &str, + provider_state: &str, + state: agent_task_service::CookContinuationState, +) -> Value { + let (status, guidance) = match state { + agent_task_service::CookContinuationState::Pending => ( + "continuation_pending", + serde_json::json!({ + "action": "await_continuation_claim", + "command": format!("homeboy agent-task cook-continue {run_id}"), + }), + ), + agent_task_service::CookContinuationState::Claimed => ( + "continuation_in_progress", + serde_json::json!({ + "action": "await_claimed_continuation", + "command": format!("homeboy agent-task status {run_id} --full"), + }), + ), + agent_task_service::CookContinuationState::Failed => ( + "continuation_recovery_required", + serde_json::json!({ + "action": "rearm_failed_continuation", + "command": format!("homeboy agent-task cook-continue {run_id} --rearm"), + }), + ), + agent_task_service::CookContinuationState::Completed => ( + "continuation_completed", + serde_json::json!({ + "action": "inspect_completed_cook", + "command": format!("homeboy agent-task status {run_id} --full"), + }), + ), + agent_task_service::CookContinuationState::Absent => ( + "continuation_not_scheduled", + serde_json::json!({ + "action": "reconcile_terminal_attempt", + "command": format!("homeboy agent-task reconcile {run_id} --dry-run"), + }), + ), + }; serde_json::json!({ "schema": "homeboy/agent-task-cook/v1", "cook_id": cook_id, "latest_run_id": run_id, - "status": "continuation_pending", + "status": status, "provider": { "state": provider_state, "run_id": run_id }, "remaining_phases": ["harvest", "review", "gates", "promotion", "finalization"], - "continuation_command": format!("homeboy agent-task cook-continue {run_id}"), + "guidance": guidance, }) } -fn cook_continuation_status(cook_id: &str, run_id: &str, provider_state: &str) -> Value { +fn cook_continuation_status( + cook_id: &str, + record: &agent_task_lifecycle::AgentTaskRunRecord, +) -> Value { + let run_id = &record.run_id; + let runner_id = record.runner_id(); + let waiting_for_capacity = record + .metadata + .pointer("/runner_queue/state") + .and_then(Value::as_str) + == Some("waiting_for_capacity"); + let runner_disconnected = record + .metadata + .get("runner_liveness") + .and_then(Value::as_str) + == Some("disconnected"); + let (status, guidance) = if runner_disconnected { + ( + "recovery_required", + serde_json::json!({ + "action": "reconcile_runner_authority", + "command": format!("homeboy agent-task reconcile {run_id} --dry-run"), + "message": "The runner could not be reached during bounded reconciliation; inspect the scoped recovery before retrying provider work." + }), + ) + } else if waiting_for_capacity + || record.state == agent_task_lifecycle::AgentTaskRunState::Queued + { + let command = runner_id + .map(|runner_id| format!("homeboy runner status {runner_id}")) + .unwrap_or_else(|| format!("homeboy agent-task run {run_id}")); + ( + "accepted_unscheduled", + serde_json::json!({ + "action": if runner_id.is_some() { "await_runner_capacity" } else { "schedule_queued_run" }, + "command": command, + "message": if runner_id.is_some() { + "The runner accepted this Cook and owns its FIFO queue entry; it will schedule provider execution when a capacity lease is available." + } else { + "This Cook is durably queued but has no runner or provider boundary; schedule its queued run before expecting provider work." + } + }), + ) + } else { + ( + "observation_in_progress", + serde_json::json!({ + "action": "watch_provider", + "command": format!("homeboy agent-task logs {run_id}"), + "message": "The runner reports active provider work; follow the durable observation until its terminal result is projected." + }), + ) + }; serde_json::json!({ "schema": "homeboy/agent-task-cook/v1", "cook_id": cook_id, "latest_run_id": run_id, - "status": "in_flight", - "provider": { "state": provider_state, "run_id": run_id }, + "status": status, + "provider": { + "state": format!("{:?}", record.state), + "run_id": run_id, + "runner_id": runner_id, + "runner_job_status": record.metadata.get("runner_job_status"), + }, "remaining_phases": ["harvest", "review", "gates", "promotion", "finalization"], - "continuation_command": format!("homeboy agent-task cook-continue {run_id}"), + "guidance": guidance, }) } @@ -1614,7 +1734,7 @@ pub(super) fn retry(args: RetryArgs) -> CmdResult { #[cfg(test)] mod tests { use super::{ - cook_report_with_continuation, cook_resolved_policy_disclosure, + cook_continuation_status, cook_report_with_continuation, cook_resolved_policy_disclosure, durable_cook_identity_lines, preflight_continue_cook, }; use crate::commands::agent_task::args::CookContinueArgs; @@ -1675,6 +1795,104 @@ mod tests { ); } + #[test] + fn cook_continue_reports_runner_capacity_without_offering_a_duplicate_dispatch() { + crate::test_support::with_isolated_home(|_| { + let plan = homeboy::agents::agent_tasks::scheduler::AgentTaskPlan::new("plan", vec![]); + homeboy::agents::agent_tasks::lifecycle::submit_plan( + &plan, + Some("cook-queue-attempt-1"), + ) + .expect("persist queued attempt"); + homeboy::agents::agent_tasks::lifecycle::rewrite_record_for_test( + "cook-queue-attempt-1", + |record| { + record.metadata = serde_json::json!({ + "runner_id": "fixture-lab", + "runner_job_status": "queued", + "runner_queue": { "state": "waiting_for_capacity" }, + }); + }, + ) + .expect("record accepted queue ownership"); + let record = homeboy::agents::agent_tasks::lifecycle::status("cook-queue-attempt-1") + .expect("read queued attempt"); + + let report = cook_continuation_status("cook-queue", &record); + + assert_eq!(report["status"], "accepted_unscheduled"); + assert_eq!(report["guidance"]["action"], "await_runner_capacity"); + assert_eq!( + report["guidance"]["command"], + "homeboy runner status fixture-lab" + ); + assert!(report.get("continuation_command").is_none()); + }); + } + + #[test] + fn cook_continue_reports_a_queued_record_without_runner_as_unscheduled() { + crate::test_support::with_isolated_home(|_| { + let plan = homeboy::agents::agent_tasks::scheduler::AgentTaskPlan::new("plan", vec![]); + homeboy::agents::agent_tasks::lifecycle::submit_plan( + &plan, + Some("cook-unassigned-attempt-1"), + ) + .expect("persist queued attempt"); + let record = + homeboy::agents::agent_tasks::lifecycle::status("cook-unassigned-attempt-1") + .expect("read queued attempt without a runner boundary"); + + let report = cook_continuation_status("cook-unassigned", &record); + + assert_eq!(report["status"], "accepted_unscheduled"); + assert_eq!(report["guidance"]["action"], "schedule_queued_run"); + assert_eq!( + report["guidance"]["command"], + "homeboy agent-task run cook-unassigned-attempt-1" + ); + assert!(report.get("continuation_command").is_none()); + }); + } + + #[test] + fn cook_continue_reports_active_runner_work_with_a_watch_command() { + crate::test_support::with_isolated_home(|_| { + let plan = homeboy::agents::agent_tasks::scheduler::AgentTaskPlan::new("plan", vec![]); + homeboy::agents::agent_tasks::lifecycle::submit_plan( + &plan, + Some("cook-active-attempt-1"), + ) + .expect("persist active attempt"); + homeboy::agents::agent_tasks::lifecycle::mark_running("cook-active-attempt-1") + .expect("runner starts the provider attempt"); + homeboy::agents::agent_tasks::lifecycle::rewrite_record_for_test( + "cook-active-attempt-1", + |record| { + record.metadata = serde_json::json!({ + "runner_id": "fixture-lab", + "runner_job_status": "running", + "provider_run_ids": ["fixture-provider-run-1"], + "phase": "provider_execution", + }); + }, + ) + .expect("record active runner and provider evidence"); + let record = homeboy::agents::agent_tasks::lifecycle::status("cook-active-attempt-1") + .expect("read active runner-backed provider attempt"); + + let report = cook_continuation_status("cook-active", &record); + + assert_eq!(report["status"], "observation_in_progress"); + assert_eq!(report["guidance"]["action"], "watch_provider"); + assert_eq!( + report["guidance"]["command"], + "homeboy agent-task logs cook-active-attempt-1" + ); + assert!(report.get("continuation_command").is_none()); + }); + } + #[test] fn cook_continue_preflight_reports_a_read_only_rejection_without_creating_a_run() { crate::test_support::with_isolated_home(|_| { @@ -1683,6 +1901,7 @@ mod tests { let (report, exit_code) = preflight_continue_cook(CookContinueArgs { cook_or_attempt_id: "missing-cook".to_string(), preflight: true, + rearm: false, full: false, }) .expect("preflight returns a machine-readable rejection"); diff --git a/crates/homeboy-cli/src/commands/agent_task/tests/lifecycle.rs b/crates/homeboy-cli/src/commands/agent_task/tests/lifecycle.rs index 359e2be3e..d04b512f4 100644 --- a/crates/homeboy-cli/src/commands/agent_task/tests/lifecycle.rs +++ b/crates/homeboy-cli/src/commands/agent_task/tests/lifecycle.rs @@ -7,6 +7,7 @@ use homeboy::agents::agent_task_service::{ }; use homeboy::core::{Error, Result}; use sha2::{Digest, Sha256}; +use std::os::unix::fs::PermissionsExt; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use crate::cli_surface::{Cli, Commands}; @@ -389,6 +390,244 @@ fn cook_runner_preflight_failure_is_visible_and_resumable_through_public_command }); } +#[test] +fn cook_continue_reconciles_a_delayed_runner_attempt_then_advances_its_terminal_recipe_once() { + with_temp_home(|| { + let workspace = tempfile::tempdir().expect("workspace"); + init_runtime_component_checkout(workspace.path()); + let provider = workspace.path().join("worktree-provider.sh"); + std::fs::write( + &provider, + format!( + "#!/bin/sh\nprintf '%s\\n' '{{\"worktrees\":[{{\"handle\":\"fixture@delayed\",\"path\":\"{}\",\"branch\":\"main\",\"safety\":{{\"dirty\":false,\"unpushed\":false,\"primary\":false}}}}]}}'\n", + workspace.path().display() + ), + ) + .expect("write worktree provider"); + let mut permissions = std::fs::metadata(&provider) + .expect("worktree provider metadata") + .permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&provider, permissions) + .expect("make worktree provider executable"); + let mut config = homeboy::core::defaults::load_config(); + config.worktree_providers.insert( + "fixture".to_string(), + homeboy::core::defaults::WorktreeProviderConfig { + enabled: true, + kind: homeboy::core::defaults::WorktreeProviderKind::Command, + apply_enabled: true, + lookup_timeout_ms: 10_000, + lookup_output_limit_bytes: 64 * 1024, + commands: homeboy::core::defaults::WorktreeProviderCommands { + resolve: Some(vec![ + provider.display().to_string(), + "resolve".to_string(), + "{handle}".to_string(), + ]), + ..Default::default() + }, + list_result_mapping: Some( + homeboy::core::defaults::WorktreeProviderListResultMapping { + items: "$.worktrees".to_string(), + handle: "$.handle".to_string(), + path: "$.path".to_string(), + branch: "$.branch".to_string(), + dirty: "$.safety.dirty".to_string(), + unpushed: "$.safety.unpushed".to_string(), + primary: "$.safety.primary".to_string(), + task_url: None, + }, + ), + }, + ); + homeboy::core::defaults::save_config(&config).expect("save worktree provider config"); + let promotion_count = workspace.path().join("promotion-count"); + let promotion_provider = workspace.path().join("promotion-provider.sh"); + let patch = workspace.path().join("delayed-provider.patch"); + std::fs::write( + &promotion_provider, + format!( + "#!/bin/sh\nset -eu\ncat >/dev/null\nprintf '1\\n' >> {}\ngit -C {} apply {}\nprintf '%s\\n' '{{\"schema\":\"homeboy/agent-task-promotion-apply-response/v1\",\"workspace_path\":\"{}\"}}'\n", + promotion_count.display(), + workspace.path().display(), + patch.display(), + workspace.path().display(), + ), + ) + .expect("write deterministic promotion provider"); + let cook_id = "cook-continue-delayed"; + let run_id = "cook-continue-delayed-attempt-1"; + let plan = AgentTaskPlan::new( + "cook-continue-delayed-plan", + vec![serde_json::from_value(json!({ + "task_id": "provider", + "executor": { "backend": "fixture", "model": "fixture-model" }, + "instructions": "complete the delayed provider attempt", + "workspace": { "root": workspace.path() } + })) + .expect("provider task")], + ); + let options = homeboy::agents::agent_task_service::AgentTaskCookServiceOptions { + cook_id: cook_id.to_string(), + initial_run_id: run_id.to_string(), + initial_plan: plan.clone(), + to_worktree: "fixture@delayed".to_string(), + source_worktree_path: Some(workspace.path().to_path_buf()), + provider_command: None, + provider_invocation: Some(homeboy::core::command_invocation::CommandInvocation { + argv: vec!["sh".to_string(), promotion_provider.display().to_string()], + ..Default::default() + }), + gates: Default::default(), + max_attempts: 1, + no_finalize: true, + draft_pr: false, + base: "main".to_string(), + task_base_sha: None, + head: None, + title: "Delayed Cook continuation".to_string(), + commit_message: "Delayed Cook continuation".to_string(), + source_refs: Vec::new(), + protected_branches: Vec::new(), + ai_tool: "fixture".to_string(), + ai_model: Some("fixture-model".to_string()), + ai_used_for: "test".to_string(), + attempt_dispatcher: None, + harvest_context: homeboy::agents::agent_task_scheduler::HarvestExecutionContext::from_current_process() + .expect("harvest context"), + }; + homeboy::agents::agent_task_service::persist_initial_recipe(&options) + .expect("persist immutable recipe"); + agent_task_lifecycle::submit_plan(&plan, Some(run_id)).expect("persist provider attempt"); + agent_task_lifecycle::record_cook_attempt(cook_id, 1, run_id) + .expect("bind immutable Cook attempt"); + let executor = CountingCookExecutor::default(); + let before = continue_cook_with( + CookContinueArgs { + cook_or_attempt_id: cook_id.to_string(), + preflight: false, + rearm: false, + full: true, + }, + executor.clone(), + |_| Ok(None), + ) + .expect("cook-continue observes the queued attempt before provider scheduling"); + assert_eq!(before.0["status"], "accepted_unscheduled"); + assert_eq!(before.0["guidance"]["action"], "schedule_queued_run"); + assert_eq!( + before.0["guidance"]["command"], + format!("homeboy agent-task run {run_id}") + ); + assert_eq!(executor.executions.load(Ordering::SeqCst), 0); + + let patch_contents = "diff --git a/delayed-provider.txt b/delayed-provider.txt\nnew file mode 100644\nindex 0000000..e69de29\n--- /dev/null\n+++ b/delayed-provider.txt\n@@ -0,0 +1 @@\n+completed after runner reconciliation\n"; + std::fs::write(&patch, patch_contents).expect("write delayed provider patch"); + let patch_sha256 = format!("{:x}", Sha256::digest(patch_contents.as_bytes())); + agent_task_lifecycle::record_run_aggregate( + run_id, + &plan, + &AgentTaskAggregate { + schema: "homeboy/agent-task-aggregate/v1".to_string(), + plan_id: plan.plan_id.clone(), + status: + homeboy::agents::agent_tasks::scheduler::AgentTaskAggregateStatus::Succeeded, + totals: Default::default(), + outcomes: vec![AgentTaskOutcome { + schema: AGENT_TASK_OUTCOME_SCHEMA.to_string(), + task_id: "provider".to_string(), + status: AgentTaskOutcomeStatus::Succeeded, + summary: Some("delayed provider completed".to_string()), + failure_classification: None, + artifacts: vec![AgentTaskArtifact { + id: "delayed-provider-patch".to_string(), + kind: "patch".to_string(), + path: Some(patch.display().to_string()), + size_bytes: Some(patch_contents.len() as u64), + sha256: Some(patch_sha256), + ..Default::default() + }], + typed_artifacts: Vec::new(), + evidence_refs: Vec::new(), + diagnostics: Vec::new(), + outputs: Value::Null, + workflow: None, + follow_up: None, + metadata: Value::Null, + }], + events: Vec::new(), + artifact_lineage: Vec::new(), + child_runs: Vec::new(), + artifact_bindings: Vec::new(), + queue: Default::default(), + }, + ) + .expect("publish delayed provider aggregate"); + + let after = continue_cook_with( + CookContinueArgs { + cook_or_attempt_id: cook_id.to_string(), + preflight: false, + rearm: false, + full: true, + }, + executor.clone(), + |_| Ok(None), + ) + .expect("the same cook-continue advances the terminal attempt"); + assert_eq!( + after.1, 1, + "the deterministic promotion fixture surfaces its declared gate failure" + ); + assert_eq!(std::fs::read_to_string(&promotion_count).unwrap(), "1\n"); + assert_eq!(after.0["failure_context"]["phase"], "promotion"); + assert_ne!(after.0["status"], "observation_in_progress"); + assert_eq!(executor.executions.load(Ordering::SeqCst), 0); + + let replay = continue_cook_with( + CookContinueArgs { + cook_or_attempt_id: cook_id.to_string(), + preflight: false, + rearm: false, + full: true, + }, + executor.clone(), + |_| Ok(None), + ) + .expect("replaying cook-continue is idempotent"); + assert_eq!(replay.1, 0, "the failed continuation claim is not replayed"); + assert_eq!(replay.0["status"], "continuation_recovery_required"); + assert_eq!( + replay.0["guidance"]["command"], + format!("homeboy agent-task cook-continue {run_id} --rearm") + ); + assert_eq!( + std::fs::read_to_string(&promotion_count).unwrap(), + "1\n", + "a failed continuation never silently replays its promotion provider" + ); + let rearmed = continue_cook_with( + CookContinueArgs { + cook_or_attempt_id: cook_id.to_string(), + preflight: false, + rearm: true, + full: true, + }, + executor.clone(), + |_| Ok(None), + ) + .expect("an explicit rearm may retry the failed continuation"); + assert_eq!(rearmed.1, 1); + assert_eq!( + std::fs::read_to_string(&promotion_count).unwrap(), + "1\n", + "explicit rearm resumes the durable promotion result without reapplying its provider side effect" + ); + assert_eq!(executor.executions.load(Ordering::SeqCst), 0); + }); +} + #[test] fn controller_proxy_status_and_logs_resolve_before_runner_child_is_known() { with_temp_home(|| { diff --git a/crates/homeboy-cli/src/commands/agent_task/tests/support.rs b/crates/homeboy-cli/src/commands/agent_task/tests/support.rs index c0847b9c2..a81a30c2a 100644 --- a/crates/homeboy-cli/src/commands/agent_task/tests/support.rs +++ b/crates/homeboy-cli/src/commands/agent_task/tests/support.rs @@ -8,6 +8,7 @@ pub(crate) use std::process::Command; pub(in crate::commands::agent_task) use super::super::super::agent_task_dispatch::{ DispatchArgs, DispatchCoreArgs, }; +pub(in crate::commands::agent_task) use super::super::args::CookContinueArgs; pub(in crate::commands::agent_task) use super::super::args::{ AgentTaskControllerApplyEventArgs, AgentTaskControllerDispatchArgs, AgentTaskControllerFromSpecArgs, AgentTaskControllerMaterializeArgs, @@ -23,9 +24,10 @@ pub(in crate::commands::agent_task) use super::super::controller::{ controller_run_from_spec_with_test_executor, controller_run_next_with_executor, }; pub(in crate::commands::agent_task) use super::super::run::{ - resume, retry, run_cook_with_executor, run_cook_with_executor_and_dispatcher, run_loaded_plan, - run_next_with_executor, run_resume_with_executor, run_submitted, run_submitted_with_executor, - submit, validate_cook_request, + continue_cook_with, resume, retry, run_cook_with_executor, + run_cook_with_executor_and_dispatcher, run_loaded_plan, run_next_with_executor, + run_resume_with_executor, run_submitted, run_submitted_with_executor, submit, + validate_cook_request, }; pub(in crate::commands::agent_task) use super::super::status::{ cancel, diagnose, evidence, logs, replay_provider_boundary, status,