//! Split from `agent_task_controller_service` god file (#5208). Structural move only.
#![allow(unused_imports)]
use super::*;
use std::fs;
use std::io::{self, Read};
use std::path::PathBuf;
use std::process::Stdio;
use std::thread;
use std::time::{Duration, Instant};

const COMMAND_GATE_DEFAULT_TIMEOUT_SECONDS: u64 = 300;
const RUN_COMMAND_DEFAULT_TIMEOUT_SECONDS: u64 = 900;
const COMMAND_GATE_OUTPUT_CAP_BYTES: usize = 64 * 1024;

pub(super) fn execute_controller_action<E, D>(
    record: &mut AgentTaskLoopControllerRecord,
    action_id: &str,
    executor: E,
    dispatch: &D,
) -> Result<AgentTaskRunResult<ControllerActionReport>>
where
    E: AgentTaskExecutorAdapter + Clone,
    D: ControllerDispatchHook,
{
    execute_controller_action_with_runner_availability(
        record,
        action_id,
        executor,
        dispatch,
        controller_runner_availability,
    )
}

pub(super) fn execute_controller_action_with_runner_availability<E, D, F>(
    record: &mut AgentTaskLoopControllerRecord,
    action_id: &str,
    executor: E,
    dispatch: &D,
    mut runner_availability: F,
) -> Result<AgentTaskRunResult<ControllerActionReport>>
where
    E: AgentTaskExecutorAdapter + Clone,
    D: ControllerDispatchHook,
    F: FnMut(&str) -> AgentTaskLoopRunnerAvailability,
{
    if let Some(result) = enforce_controller_action_cap(record)? {
        return Ok(result);
    }

    let action = claim_controller_action(record, action_id)?;
    controller::write_controller(record)?;

    if let Some(result) =
        enforce_claimed_action_runner_policy(record, action_id, &action, &mut runner_availability)?
    {
        return Ok(result);
    }

    match execute_claimed_controller_action(record, &action, executor, dispatch) {
        Ok((execution, exit_code)) => {
            let mut exit_code = exit_code;
            let missing_required_artifacts = if exit_code == 0 {
                validate_required_action_artifacts(&action, &execution)
            } else {
                Vec::new()
            };
            if missing_required_artifacts.is_empty() {
                complete_controller_action(record, action_id, &execution, exit_code)?;
            } else {
                exit_code = 1;
                fail_controller_action_with_diagnostics(
                    record,
                    action_id,
                    required_artifact_diagnostics(&missing_required_artifacts),
                    &execution,
                )?;
            }
            controller::write_controller(record)?;
            let failure_summary = (exit_code != 0)
                .then(|| controller_action_failure_summary(record, &action, &execution));
            Ok(AgentTaskRunResult {
                value: ControllerActionReport {
                    schema: ACTION_RESULT_SCHEMA,
                    loop_id: record.loop_id.clone(),
                    claimed: true,
                    action_id: Some(action_id.to_string()),
                    status: Some(
                        if exit_code == 0 {
                            "completed"
                        } else {
                            "failed"
                        }
                        .to_string(),
                    ),
                    failure_summary,
                    runtime_evidence: controller_action_runtime_evidence(&execution),
                    execution: Some(execution),
                    controller: record.clone(),
                },
                exit_code,
            })
        }
        Err(error) => {
            fail_controller_action(record, action_id, &error.to_string())?;
            controller::write_controller(record)?;
            Err(error)
        }
    }
}

/// Guard against unbounded controller growth. When the controller has
/// accumulated at least [`MAX_CONTROLLER_LIFETIME_ACTIONS`] actions, escalate to
/// a terminal state instead of claiming and executing (and thereby recording)
/// yet more actions. Mirrors the deterministic loop's max-iteration guard, which
/// transitions to a blocked terminal status once the iteration ceiling is hit.
fn enforce_controller_action_cap(
    record: &mut AgentTaskLoopControllerRecord,
) -> Result<Option<AgentTaskRunResult<ControllerActionReport>>> {
    let action_count = record.next_actions.len();
    if action_count < MAX_CONTROLLER_LIFETIME_ACTIONS {
        return Ok(None);
    }

    let reason = format!(
        "controller accumulated {action_count} actions, reaching the lifetime cap of {MAX_CONTROLLER_LIFETIME_ACTIONS}; escalating to prevent unbounded loop growth"
    );

    // Record the terminal outcome and escalate only once: repeated `run`/
    // `resume` invocations against an already-capped controller must remain
    // idempotent and must not keep appending terminal outcomes.
    if record.state != AgentTaskLoopControllerState::Escalated {
        record.record_terminal_outcome(
            AgentTaskLoopTerminalStatus::Failed,
            reason.clone(),
            None,
            None,
            serde_json::json!({
                "mode": "controller_action_cap",
                "action_count": action_count,
                "max_actions": MAX_CONTROLLER_LIFETIME_ACTIONS,
            }),
        );
        record.state = AgentTaskLoopControllerState::Escalated;
        controller::write_controller(record)?;
    }

    let execution = serde_json::json!({
        "mode": "controller_action_cap",
        "status": "escalated",
        "action_count": action_count,
        "max_actions": MAX_CONTROLLER_LIFETIME_ACTIONS,
        "reason": reason,
    });
    Ok(Some(AgentTaskRunResult {
        value: ControllerActionReport {
            schema: ACTION_RESULT_SCHEMA,
            loop_id: record.loop_id.clone(),
            claimed: false,
            action_id: None,
            status: Some("escalated".to_string()),
            failure_summary: None,
            runtime_evidence: controller_action_runtime_evidence(&execution),
            execution: Some(execution),
            controller: record.clone(),
        },
        exit_code: 1,
    }))
}

fn enforce_claimed_action_runner_policy<F>(
    record: &mut AgentTaskLoopControllerRecord,
    action_id: &str,
    action: &AgentTaskLoopPolicyActionRecord,
    runner_availability: &mut F,
) -> Result<Option<AgentTaskRunResult<ControllerActionReport>>>
where
    F: FnMut(&str) -> AgentTaskLoopRunnerAvailability,
{
    let decision = record.resolve_action_runner_policy(&action.action, runner_availability);
    if let Some(blocked_status) = decision.blocked_status {
        let diagnostic = decision.diagnostic.ok_or_else(|| {
            Error::internal_unexpected(
                "runner policy returned a blocked status without a diagnostic".to_string(),
            )
        })?;
        record.block_action_for_runner_policy(action_id, blocked_status, diagnostic.clone())?;
        controller::write_controller(record)?;
        let execution = serde_json::json!({
            "mode": "runner_policy",
            "status": action_status_report_label(blocked_status),
            "diagnostics": [diagnostic],
        });
        return Ok(Some(AgentTaskRunResult {
            value: ControllerActionReport {
                schema: ACTION_RESULT_SCHEMA,
                loop_id: record.loop_id.clone(),
                claimed: true,
                action_id: Some(action_id.to_string()),
                status: Some(action_status_report_label(blocked_status).to_string()),
                failure_summary: Some(controller_action_failure_summary(
                    record, action, &execution,
                )),
                runtime_evidence: controller_action_runtime_evidence(&execution),
                execution: Some(execution),
                controller: record.clone(),
            },
            exit_code: 1,
        }));
    }

    if let Some(AgentTaskLoopRunnerExecutionTarget::Runner(runner)) = decision.target {
        let diagnostic = AgentTaskLoopActionDiagnostic {
            code: "runner_action_dispatch_unimplemented".to_string(),
            message: format!(
                "controller action '{}' targets runner `{runner}`, but controller runner dispatch is not implemented; refusing local fallback",
                action.action_id
            ),
            runner: Some(runner.clone()),
            details: serde_json::json!({
                "action_id": action.action_id,
                "dedupe_key": action.dedupe_key,
            }),
        };
        let execution = serde_json::json!({
            "mode": "runner_policy",
            "status": "failed",
            "target": "runner",
            "runner": runner,
            "diagnostics": [diagnostic],
        });
        complete_controller_action(record, action_id, &execution, 1)?;
        controller::write_controller(record)?;
        return Ok(Some(AgentTaskRunResult {
            value: ControllerActionReport {
                schema: ACTION_RESULT_SCHEMA,
                loop_id: record.loop_id.clone(),
                claimed: true,
                action_id: Some(action_id.to_string()),
                status: Some("failed".to_string()),
                failure_summary: Some(controller_action_failure_summary(
                    record, action, &execution,
                )),
                runtime_evidence: controller_action_runtime_evidence(&execution),
                execution: Some(execution),
                controller: record.clone(),
            },
            exit_code: 1,
        }));
    }

    Ok(None)
}

fn controller_runner_availability(runner_id: &str) -> AgentTaskLoopRunnerAvailability {
    match runner::status(runner_id) {
        Ok(status)
            if status.connected
                && status.stale_daemon.is_none()
                && status.active_job_state == RunnerActiveJobState::Available =>
        {
            AgentTaskLoopRunnerAvailability::Available
        }
        Ok(status) => AgentTaskLoopRunnerAvailability::Unavailable {
            reason: format!(
                "runner `{runner_id}` is not available for controller action execution: state={:?}, connected={}, stale_daemon={}, active_job_state={:?}",
                status.state,
                status.connected,
                status.stale_daemon.is_some(),
                status.active_job_state
            ),
        },
        Err(error) => AgentTaskLoopRunnerAvailability::Unavailable {
            reason: format!("runner `{runner_id}` is not available for controller action execution: {error}"),
        },
    }
}

fn action_status_report_label(status: AgentTaskLoopActionStatus) -> &'static str {
    match status {
        AgentTaskLoopActionStatus::Pending => "pending",
        AgentTaskLoopActionStatus::Running => "running",
        AgentTaskLoopActionStatus::AlreadySatisfied => "already_satisfied",
        AgentTaskLoopActionStatus::Completed => "completed",
        AgentTaskLoopActionStatus::Failed => "failed",
        AgentTaskLoopActionStatus::BlockedRunnerUnavailable => "blocked_runner_unavailable",
        AgentTaskLoopActionStatus::BlockedRemoteMaterialization => "blocked_remote_materialization",
        AgentTaskLoopActionStatus::BlockedLocalFallbackDenied => "blocked_local_fallback_denied",
    }
}

fn controller_action_failure_summary(
    record: &AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    execution: &Value,
) -> ControllerActionFailureSummary {
    let request = action_request(&action.action);
    let context = find_failure_context(execution).unwrap_or_default();
    ControllerActionFailureSummary {
        action_id: action.action_id.clone(),
        dedupe_key: action.dedupe_key.clone(),
        workflow_id: request.and_then(workflow_id_from_request),
        run_id: request
            .and_then(|request| string_field(request, "run_id"))
            .or_else(|| find_string_field(execution, "run_id")),
        task_id: context
            .task_id
            .or_else(|| find_string_field(execution, "task_id")),
        phase: Some(record.phase.clone()),
        provider: context.provider,
        failure_phase: context.failure_phase,
        runtime_context: context.runtime_context,
        replay_command: context.replay_command,
        diagnostic: context
            .diagnostic
            .or_else(|| first_action_diagnostic_message(record, &action.action_id))
            .unwrap_or_else(|| "controller action failed".to_string()),
    }
}

fn action_request(action: &AgentTaskLoopPolicyAction) -> Option<&Value> {
    match action {
        AgentTaskLoopPolicyAction::SpawnTask { request, .. }
        | AgentTaskLoopPolicyAction::SpawnController { request, .. }
        | AgentTaskLoopPolicyAction::SpawnSubloop { request, .. } => Some(request),
        AgentTaskLoopPolicyAction::FanOut {
            request_template, ..
        }
        | AgentTaskLoopPolicyAction::RouteFinding {
            request_template, ..
        } => Some(request_template),
        _ => None,
    }
}

fn workflow_id_from_request(request: &Value) -> Option<String> {
    let context = request.pointer("/dispatch/client_context")?;
    if let Some(workflow_id) = context.get("workflow_id").and_then(Value::as_str) {
        return Some(workflow_id.to_string());
    }
    let context = context.as_str()?;
    let parsed: Value = serde_json::from_str(context).ok()?;
    string_field(&parsed, "workflow_id")
}

fn first_action_diagnostic_message(
    record: &AgentTaskLoopControllerRecord,
    action_id: &str,
) -> Option<String> {
    record
        .next_actions
        .iter()
        .find(|candidate| candidate.action_id == action_id)?
        .diagnostics
        .first()
        .map(|diagnostic| diagnostic.message.clone())
}

#[derive(Default)]
struct FailureContext {
    diagnostic: Option<String>,
    task_id: Option<String>,
    provider: Option<String>,
    failure_phase: Option<String>,
    runtime_context: Option<Value>,
    replay_command: Option<String>,
}

fn find_failure_context(value: &Value) -> Option<FailureContext> {
    match value {
        Value::Object(map) => {
            if let Some(Value::Array(diagnostics)) = map.get("diagnostics") {
                if let Some(diagnostic) = diagnostics.iter().find_map(Value::as_object) {
                    let message = diagnostic
                        .get("message")
                        .and_then(Value::as_str)
                        .map(str::to_string);
                    if message.is_some() {
                        return Some(FailureContext {
                            diagnostic: message,
                            task_id: string_field(value, "task_id"),
                            provider: diagnostic
                                .get("provider")
                                .and_then(Value::as_str)
                                .map(str::to_string)
                                .or_else(|| nested_string_field(diagnostic, "data", "provider"))
                                .or_else(|| nested_string_field(diagnostic, "data", "provider_id")),
                            failure_phase: diagnostic
                                .get("phase")
                                .and_then(Value::as_str)
                                .map(str::to_string)
                                .or_else(|| nested_string_field(diagnostic, "data", "phase"))
                                .or_else(|| {
                                    diagnostic
                                        .get("class")
                                        .and_then(Value::as_str)
                                        .map(str::to_string)
                                }),
                            runtime_context: diagnostic
                                .get("data")
                                .and_then(|data| data.get("runtime_context"))
                                .cloned()
                                .or_else(|| find_value_field(value, "runtime_context")),
                            replay_command: diagnostic
                                .get("data")
                                .and_then(|data| string_field(data, "replay_command"))
                                .or_else(|| find_string_field(value, "replay_command")),
                        });
                    }
                }
            }
            map.values().find_map(find_failure_context)
        }
        Value::Array(items) => items.iter().find_map(find_failure_context),
        _ => None,
    }
}

fn string_field(value: &Value, field: &str) -> Option<String> {
    value.get(field).and_then(Value::as_str).map(str::to_string)
}

fn nested_string_field(
    map: &serde_json::Map<String, Value>,
    parent: &str,
    field: &str,
) -> Option<String> {
    map.get(parent)?.get(field)?.as_str().map(str::to_string)
}

fn find_string_field(value: &Value, field: &str) -> Option<String> {
    match value {
        Value::Object(map) => map
            .get(field)
            .and_then(Value::as_str)
            .map(str::to_string)
            .or_else(|| {
                map.values()
                    .find_map(|value| find_string_field(value, field))
            }),
        Value::Array(items) => items
            .iter()
            .find_map(|value| find_string_field(value, field)),
        _ => None,
    }
}

fn find_value_field(value: &Value, field: &str) -> Option<Value> {
    match value {
        Value::Object(map) => map.get(field).cloned().or_else(|| {
            map.values()
                .find_map(|value| find_value_field(value, field))
        }),
        Value::Array(items) => items
            .iter()
            .find_map(|value| find_value_field(value, field)),
        _ => None,
    }
}

pub(super) fn execute_claimed_controller_action<E, D>(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    executor: E,
    dispatch: &D,
) -> Result<(Value, i32)>
where
    E: AgentTaskExecutorAdapter + Clone,
    D: ControllerDispatchHook,
{
    match &action.action {
        AgentTaskLoopPolicyAction::SpawnTask {
            dedupe_key,
            entity_id,
            request,
        } => execute_spawn_task_action(
            record,
            action,
            dedupe_key,
            entity_id.as_deref(),
            request,
            executor,
            dispatch,
        ),
        AgentTaskLoopPolicyAction::RunCommand {
            dedupe_key,
            entity_id,
            request,
        } => execute_run_command_action(record, action, dedupe_key, entity_id.as_deref(), request),
        AgentTaskLoopPolicyAction::FanOut {
            dedupe_key,
            entity_ids,
            dynamic_artifact,
            group_by,
            requires_non_empty,
            max_items,
            fail_fast,
            request_template,
        } => execute_fan_out_action(
            record,
            action,
            dedupe_key,
            entity_ids,
            dynamic_artifact.as_deref(),
            group_by,
            *requires_non_empty,
            *max_items,
            *fail_fast,
            request_template,
            executor,
            dispatch,
        ),
        AgentTaskLoopPolicyAction::SpawnController {
            dedupe_key,
            loop_id,
            entity_id,
            phase,
            config_version,
            request,
        } => execute_spawn_controller_action(
            record,
            action,
            "spawn_controller",
            dedupe_key,
            loop_id,
            entity_id.as_deref(),
            phase,
            config_version,
            request,
        ),
        AgentTaskLoopPolicyAction::SpawnSubloop {
            dedupe_key,
            loop_id,
            entity_id,
            phase,
            config_version,
            request,
        } => execute_spawn_controller_action(
            record,
            action,
            "spawn_subloop",
            dedupe_key,
            loop_id,
            entity_id.as_deref(),
            phase,
            config_version,
            request,
        ),
        AgentTaskLoopPolicyAction::Join { wait_key } => Ok((
            serde_json::json!({ "mode": "join", "wait_key": wait_key }),
            0,
        )),
        AgentTaskLoopPolicyAction::WaitForEvent(wait) => Ok((
            serde_json::json!({ "mode": "wait_for_event", "wait_key": wait.wait_key }),
            0,
        )),
        AgentTaskLoopPolicyAction::WaitForController {
            loop_id,
            entity_id,
            wait_key,
            terminal_states,
        } => execute_wait_for_controller_action(
            record,
            loop_id,
            entity_id.as_deref(),
            wait_key.as_deref(),
            terminal_states,
        ),
        AgentTaskLoopPolicyAction::RunGates {
            bundle_id,
            entity_id,
        } => execute_run_gates_action(record, bundle_id, entity_id.as_deref()),
        AgentTaskLoopPolicyAction::OwnPrUntilGreen {
            ownership,
            entity_id,
        } => execute_own_pr_until_green_action(record, ownership, entity_id.as_deref()),
        AgentTaskLoopPolicyAction::MarkHumanReady { entity_id, reason } => {
            record.mark_human_ready(entity_id, reason.clone())?;
            Ok((
                serde_json::json!({ "mode": "mark_human_ready", "entity_id": entity_id }),
                0,
            ))
        }
        AgentTaskLoopPolicyAction::Complete { reason } => {
            record.state = AgentTaskLoopControllerState::Completed;
            Ok((
                serde_json::json!({ "mode": "complete", "reason": reason }),
                0,
            ))
        }
        AgentTaskLoopPolicyAction::Abandon { reason } => {
            record.state = AgentTaskLoopControllerState::Abandoned;
            Ok((
                serde_json::json!({ "mode": "abandon", "reason": reason }),
                0,
            ))
        }
        AgentTaskLoopPolicyAction::Escalate { reason } => {
            record.state = AgentTaskLoopControllerState::Escalated;
            Ok((
                serde_json::json!({ "mode": "escalate", "reason": reason }),
                0,
            ))
        }
        AgentTaskLoopPolicyAction::RouteFinding {
            finding,
            dedupe_key,
            entity_id,
            request_template,
        } => execute_route_finding_action(
            record,
            action,
            finding,
            dedupe_key,
            entity_id.as_deref(),
            request_template,
            executor,
            dispatch,
        ),
        AgentTaskLoopPolicyAction::Retry { target_run_id } => {
            execute_retry_action(record, action, target_run_id)
        }
        AgentTaskLoopPolicyAction::RequestChanges {
            target_run_id,
            feedback_id,
        } => execute_request_changes_action(record, action, target_run_id, feedback_id.as_deref()),
    }
}

fn execute_run_command_action(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    dedupe_key: &str,
    entity_id: Option<&str>,
    request: &Value,
) -> Result<(Value, i32)> {
    let request = hydrate_consumed_artifacts(record, request);
    let request = request_with_required_workflow_artifacts(record, &request);
    let execution = request.get("execution").unwrap_or(&Value::Null);
    let command = required_string(execution, "command")?;
    let args = execution
        .get("args")
        .and_then(Value::as_array)
        .map(|args| {
            args.iter()
                .filter_map(Value::as_str)
                .map(str::to_string)
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    let required_artifacts = request
        .get("artifacts")
        .and_then(Value::as_array)
        .map(|artifacts| {
            artifacts
                .iter()
                .filter_map(Value::as_str)
                .map(str::to_string)
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    let timeout_seconds = execution
        .get("timeout_seconds")
        .and_then(Value::as_u64)
        .filter(|seconds| *seconds > 0)
        .unwrap_or(RUN_COMMAND_DEFAULT_TIMEOUT_SECONDS);

    let io_dir = loop_action_io_dir(&record.loop_id, &action.action_id)?;
    fs::create_dir_all(&io_dir).map_err(|error| Error::internal_io(error.to_string(), None))?;
    let input_path = io_dir.join("input.json");
    let output_path = io_dir.join("output.json");
    let input = serde_json::json!({
        "schema": "homeboy/agent-task-loop-command-input/v1",
        "loop_id": record.loop_id,
        "action_id": action.action_id,
        "dedupe_key": dedupe_key,
        "entity_id": entity_id,
        "request": request,
        "controller": &*record,
    });
    write_json_file(&input_path, &input)?;

    let mut process = Command::new(&command);
    process.args(&args);
    if let Some(cwd) = execution
        .get("cwd")
        .and_then(Value::as_str)
        .filter(|cwd| !cwd.is_empty())
    {
        process.current_dir(cwd);
    }
    process.env("HOMEBOY_LOOP_ACTION_INPUT", &input_path);
    process.env("HOMEBOY_LOOP_ACTION_OUTPUT", &output_path);
    process.env("HOMEBOY_LOOP_ID", &record.loop_id);
    process.env("HOMEBOY_LOOP_ACTION_ID", &action.action_id);
    process.env("HOMEBOY_LOOP_ACTION_DEDUPE_KEY", dedupe_key);

    process.stdout(Stdio::piped()).stderr(Stdio::piped());
    let output = run_controller_command_with_timeout(process, &command, timeout_seconds)?;
    let exit_code = output.exit_code.unwrap_or(1);
    let result = if output_path.exists() {
        read_json_file(&output_path)?
    } else {
        serde_json::json!({})
    };
    let artifacts = result.get("artifacts").cloned().unwrap_or(Value::Null);
    let missing = missing_required_command_artifacts(&required_artifacts, &artifacts);
    let command_success = exit_code == 0 && missing.is_empty();
    let effective_exit_code = if command_success { 0 } else { 1 };

    if command_success {
        record_run_command_outputs(record, action, dedupe_key, entity_id, &request, &artifacts)?;
    }

    Ok((
        serde_json::json!({
            "mode": "run_command",
            "command": command,
            "args": args,
            "input_path": input_path,
            "output_path": output_path,
            "timeout_seconds": timeout_seconds,
            "timed_out": output.timed_out,
            "exit_code": exit_code,
            "signal": output.signal,
            "stdout": String::from_utf8_lossy(&output.stdout.bytes),
            "stderr": String::from_utf8_lossy(&output.stderr.bytes),
            "stdout_bytes": output.stdout.total_bytes,
            "stderr_bytes": output.stderr.total_bytes,
            "stdout_stored_bytes": output.stdout.bytes.len(),
            "stderr_stored_bytes": output.stderr.bytes.len(),
            "stdout_truncated": output.stdout.truncated,
            "stderr_truncated": output.stderr.truncated,
            "missing_artifacts": missing,
            "result": result,
        }),
        effective_exit_code,
    ))
}

struct ControllerCommandOutput {
    exit_code: Option<i32>,
    signal: Option<String>,
    timed_out: bool,
    stdout: CappedCommandOutput,
    stderr: CappedCommandOutput,
}

fn run_controller_command_with_timeout(
    mut process: Command,
    command: &str,
    timeout_seconds: u64,
) -> Result<ControllerCommandOutput> {
    configure_controller_command_process_group(&mut process);
    let mut child = process.spawn().map_err(|error| {
        Error::internal_io(
            format!("failed to execute controller command '{command}': {error}"),
            None,
        )
    })?;
    let stdout = child.stdout.take().map(read_capped_command_output);
    let stderr = child.stderr.take().map(read_capped_command_output);
    let deadline = Instant::now() + Duration::from_secs(timeout_seconds);
    let mut timed_out = false;
    let exit_status = loop {
        if let Some(status) = child.try_wait().map_err(|error| {
            Error::internal_io(
                format!("failed to poll controller command '{command}': {error}"),
                None,
            )
        })? {
            break status;
        }
        if Instant::now() >= deadline {
            timed_out = true;
            kill_controller_command_process_group(&mut child);
            break child.wait().map_err(|error| {
                Error::internal_io(
                    format!("failed to reap timed out controller command '{command}': {error}"),
                    None,
                )
            })?;
        }
        thread::sleep(Duration::from_millis(25));
    };
    Ok(ControllerCommandOutput {
        exit_code: exit_status.code(),
        signal: exit_status_signal(&exit_status),
        timed_out,
        stdout: collect_capped_command_output(stdout, command, "stdout")?,
        stderr: collect_capped_command_output(stderr, command, "stderr")?,
    })
}

#[cfg(unix)]
fn configure_controller_command_process_group(process: &mut Command) {
    use std::os::unix::process::CommandExt;

    process.process_group(0);
}

#[cfg(not(unix))]
fn configure_controller_command_process_group(_process: &mut Command) {}

#[cfg(unix)]
fn kill_controller_command_process_group(child: &mut std::process::Child) {
    let pid = child.id();
    if pid > i32::MAX as u32 {
        let _ = child.kill();
        return;
    }
    let pgid = -(pid as i32);
    unsafe {
        libc::kill(pgid, libc::SIGKILL);
    }
}

#[cfg(not(unix))]
fn kill_controller_command_process_group(child: &mut std::process::Child) {
    let _ = child.kill();
}

#[cfg(unix)]
fn exit_status_signal(status: &std::process::ExitStatus) -> Option<String> {
    use std::os::unix::process::ExitStatusExt;
    status.signal().map(|signal| signal.to_string())
}

#[cfg(not(unix))]
fn exit_status_signal(_status: &std::process::ExitStatus) -> Option<String> {
    None
}

fn loop_action_io_dir(loop_id: &str, action_id: &str) -> Result<PathBuf> {
    Ok(crate::core::paths::homeboy_data()?
        .join("agent-task-loop-actions")
        .join(sanitize_loop_action_id(loop_id))
        .join(action_id))
}

fn sanitize_loop_action_id(raw: &str) -> String {
    raw.chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
                ch
            } else {
                '_'
            }
        })
        .collect()
}

fn write_json_file(path: &PathBuf, value: &Value) -> Result<()> {
    let payload = serde_json::to_string_pretty(value)
        .map_err(|error| Error::internal_json(error.to_string(), None))?;
    fs::write(path, format!("{payload}\n"))
        .map_err(|error| Error::internal_io(error.to_string(), None))
}

fn read_json_file(path: &PathBuf) -> Result<Value> {
    let payload =
        fs::read_to_string(path).map_err(|error| Error::internal_io(error.to_string(), None))?;
    serde_json::from_str(&payload).map_err(|error| Error::internal_json(error.to_string(), None))
}

fn missing_required_command_artifacts(required: &[String], artifacts: &Value) -> Vec<String> {
    required
        .iter()
        .filter(|artifact| artifacts.get(artifact.as_str()).is_none_or(Value::is_null))
        .cloned()
        .collect()
}

fn record_run_command_outputs(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    dedupe_key: &str,
    entity_id: Option<&str>,
    request: &Value,
    artifacts: &Value,
) -> Result<()> {
    let run_id = format!("{}:{}", record.loop_id, action.action_id);
    let artifact_refs = artifact_refs_from_command_result(artifacts);
    persist_controller_artifacts(&record.loop_id, &action.action_id, artifacts)?;
    if let Some(entity_id) = entity_id {
        if let Some(entity) = record.entities.get_mut(entity_id) {
            entity.artifact_refs.extend(artifact_refs.clone());
        }
    }
    if !record
        .task_lineage
        .iter()
        .any(|lineage| lineage.run_id == run_id)
    {
        record.task_lineage.push(AgentTaskLoopTaskLineage {
            run_id: run_id.clone(),
            task_id: None,
            parent_run_id: None,
            parent_task_id: None,
            entity_id: entity_id.map(str::to_string),
            dedupe_key: Some(dedupe_key.to_string()),
            artifact_refs,
            inputs: request.clone(),
            outputs: serde_json::json!({ "artifacts": artifacts }),
        });
    }
    Ok(())
}

fn persist_controller_artifacts(loop_id: &str, action_id: &str, artifacts: &Value) -> Result<()> {
    let Some(object) = artifacts.as_object() else {
        return Ok(());
    };
    if object.is_empty() {
        return Ok(());
    }
    let root = crate::core::paths::artifact_root()?
        .join("agent-task-loop-controller")
        .join(sanitize_loop_action_id(loop_id))
        .join(action_id);
    fs::create_dir_all(&root).map_err(|error| Error::internal_io(error.to_string(), None))?;
    for (artifact_id, artifact) in object {
        let path = root.join(format!("{}.json", sanitize_loop_action_id(artifact_id)));
        write_json_file(&path, artifact)?;
    }
    Ok(())
}

fn artifact_refs_from_command_result(artifacts: &Value) -> Vec<AgentTaskLoopArtifactRef> {
    let Some(object) = artifacts.as_object() else {
        return Vec::new();
    };
    object
        .iter()
        .map(|(artifact_id, artifact)| {
            let uri = artifact
                .get("artifact_url")
                .or_else(|| artifact.get("url"))
                .or_else(|| artifact.get("path"))
                .and_then(Value::as_str)
                .unwrap_or(artifact_id)
                .to_string();
            AgentTaskLoopArtifactRef {
                uri,
                kind: artifact
                    .get("schema")
                    .and_then(Value::as_str)
                    .map(str::to_string),
                role: None,
                label: Some(artifact_id.clone()),
                semantic_key: None,
            }
        })
        .collect()
}

pub(super) fn execute_retry_action(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    target_run_id: &str,
) -> Result<(Value, i32)> {
    let retry = agent_task_service::retry(target_run_id, None, false)?;
    let retry_run_id = retry.record.run_id.clone();
    if !record
        .task_lineage
        .iter()
        .any(|lineage| lineage.run_id == retry_run_id)
    {
        record.task_lineage.push(AgentTaskLoopTaskLineage {
            run_id: retry_run_id.clone(),
            task_id: None,
            parent_run_id: Some(target_run_id.to_string()),
            parent_task_id: None,
            entity_id: None,
            dedupe_key: action.dedupe_key.clone(),
            artifact_refs: Vec::new(),
            inputs: serde_json::json!({ "target_run_id": target_run_id }),
            outputs: Value::Null,
        });
    }
    push_controller_history(
        record,
        "controller.action.retry_queued",
        None,
        serde_json::json!({
            "action_id": action.action_id,
            "target_run_id": target_run_id,
            "retry_run_id": retry_run_id,
        }),
    );
    Ok((
        serde_json::json!({
            "mode": "retry",
            "target_run_id": target_run_id,
            "retry_run_id": retry.record.run_id,
            "record": retry.record,
            "run": retry.run,
        }),
        0,
    ))
}

pub(super) fn execute_request_changes_action(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    target_run_id: &str,
    feedback_id: Option<&str>,
) -> Result<(Value, i32)> {
    let feedback_id = feedback_id
        .map(str::to_string)
        .unwrap_or_else(|| format!("feedback-{}", record.feedback.len() + 1));
    let feedback = controller::AgentTaskLoopFeedbackArtifact {
        feedback_id: feedback_id.clone(),
        review_run_id: None,
        target_run_id: Some(target_run_id.to_string()),
        target_task_id: None,
        target_entity_id: None,
        status: controller::AgentTaskLoopFeedbackStatus::ChangesRequested,
        findings: Vec::new(),
        payload: serde_json::json!({
            "action_id": action.action_id,
            "target_run_id": target_run_id,
        }),
    };
    if let Some(existing) = record
        .feedback
        .iter_mut()
        .find(|existing| existing.feedback_id == feedback.feedback_id)
    {
        *existing = feedback.clone();
    } else {
        record.feedback.push(feedback.clone());
    }
    push_controller_history(
        record,
        "controller.action.changes_requested",
        None,
        serde_json::json!({
            "action_id": action.action_id,
            "target_run_id": target_run_id,
            "feedback_id": feedback_id,
        }),
    );
    Ok((
        serde_json::json!({
            "mode": "request_changes",
            "target_run_id": target_run_id,
            "feedback": feedback,
        }),
        0,
    ))
}

pub(super) fn execute_spawn_task_action<E, D>(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    dedupe_key: &str,
    entity_id: Option<&str>,
    request: &Value,
    executor: E,
    dispatch: &D,
) -> Result<(Value, i32)>
where
    E: AgentTaskExecutorAdapter + Clone,
    D: ControllerDispatchHook,
{
    let request = hydrate_consumed_artifacts(record, request);
    let request = request_with_required_workflow_artifacts(record, &request);
    let request = &request;
    let mode = request
        .get("mode")
        .and_then(Value::as_str)
        .unwrap_or("run_plan");
    match mode {
        "run_plan" => {
            let plan = plan_from_controller_request(request)?;
            let run_id =
                controller_request_run_id(request, &record.loop_id, dedupe_key, &action.action_id);
            let submitted = if lifecycle::run_record_exists(&run_id)? {
                lifecycle::status(&run_id)?
            } else {
                lifecycle::submit_plan(&plan, Some(&run_id))?
            };
            record_controller_spawn(
                record,
                action,
                dedupe_key,
                entity_id,
                &submitted.run_id,
                request,
            )?;
            let run_result = agent_task_service::run_submitted(submitted.run_id.clone(), executor)?;
            record_controller_aggregate_evidence(
                record,
                entity_id,
                &submitted.run_id,
                &run_result.value,
            )?;
            let aggregate_value = serde_json::to_value(&run_result.value)
                .map_err(|error| Error::internal_json(error.to_string(), None))?;
            Ok((
                execution_with_request_workflow_artifacts(
                    serde_json::json!({
                        "mode": mode,
                        "run_id": submitted.run_id,
                        "submitted": submitted,
                        "aggregate": aggregate_value,
                    }),
                    request,
                ),
                run_result.exit_code,
            ))
        }
        "submit" => {
            let plan = plan_from_controller_request(request)?;
            let run_id =
                controller_request_run_id(request, &record.loop_id, dedupe_key, &action.action_id);
            let submitted = lifecycle::submit_plan(&plan, Some(&run_id))?;
            record_controller_spawn(
                record,
                action,
                dedupe_key,
                entity_id,
                &submitted.run_id,
                request,
            )?;
            Ok((
                execution_with_request_workflow_artifacts(
                    serde_json::json!({
                        "mode": mode,
                        "run_id": submitted.run_id,
                        "submitted": submitted,
                    }),
                    request,
                ),
                0,
            ))
        }
        "run" => {
            let run_id = required_string(request, "run_id")?;
            record_controller_spawn(record, action, dedupe_key, entity_id, &run_id, request)?;
            let run_result = agent_task_service::run_submitted(run_id.clone(), executor)?;
            record_controller_aggregate_evidence(record, entity_id, &run_id, &run_result.value)?;
            let aggregate_value = serde_json::to_value(&run_result.value)
                .map_err(|error| Error::internal_json(error.to_string(), None))?;
            Ok((
                execution_with_request_workflow_artifacts(
                    serde_json::json!({ "mode": mode, "run_id": run_id, "aggregate": aggregate_value }),
                    request,
                ),
                run_result.exit_code,
            ))
        }
        "resume" => {
            let run_id = required_string(request, "run_id")?;
            record_controller_spawn(record, action, dedupe_key, entity_id, &run_id, request)?;
            let run_result = agent_task_service::resume(run_id.clone(), executor)?;
            record_controller_aggregate_evidence(record, entity_id, &run_id, &run_result.value)?;
            let aggregate_value = serde_json::to_value(&run_result.value)
                .map_err(|error| Error::internal_json(error.to_string(), None))?;
            Ok((
                execution_with_request_workflow_artifacts(
                    serde_json::json!({ "mode": mode, "run_id": run_id, "aggregate": aggregate_value }),
                    request,
                ),
                run_result.exit_code,
            ))
        }
        "run_next" => {
            let run_result = agent_task_service::run_next(executor)?;
            let value = match run_result.value {
                Some(aggregate) => serde_json::to_value(&aggregate)
                    .map_err(|error| Error::internal_json(error.to_string(), None))?,
                None => serde_json::json!({ "claimed": false }),
            };
            if let Some(run_id) = value.get("run_id").and_then(Value::as_str) {
                record_controller_spawn(record, action, dedupe_key, entity_id, run_id, request)?;
            }
            Ok((
                execution_with_request_workflow_artifacts(
                    serde_json::json!({ "mode": mode, "result": value }),
                    request,
                ),
                run_result.exit_code,
            ))
        }
        "dispatch" => {
            let request =
                request_with_controller_dispatch_identity(record, action, dedupe_key, request);
            let (value, exit_code) = dispatch.dispatch(&request)?;
            if let Some(run_id) = value.get("run_id").and_then(Value::as_str) {
                record_controller_spawn(record, action, dedupe_key, entity_id, run_id, &request)?;
                record_controller_result_evidence(record, entity_id, run_id, &value)?;
            }
            Ok((
                execution_with_request_workflow_artifacts(
                    serde_json::json!({ "mode": mode, "result": value }),
                    &request,
                ),
                exit_code,
            ))
        }
        other => Err(Error::validation_invalid_argument(
            "request.mode",
            format!("unsupported spawn_task request mode '{other}'"),
            Some(other.to_string()),
            Some(vec![
                "Supported modes: run_plan, submit, run, resume, run_next, dispatch".to_string(),
            ]),
        )),
    }
}

fn request_with_controller_dispatch_identity(
    record: &AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    dedupe_key: &str,
    request: &Value,
) -> Value {
    let mut request = request.clone();
    let identity = sanitize_loop_action_id(&format!(
        "{}-{}-{}",
        record.loop_id, action.action_id, dedupe_key
    ));
    let run_id = format!("controller-{identity}");
    let task_id = format!("controller-task-{identity}");

    if let Some(dispatch) = request.get_mut("dispatch").and_then(Value::as_object_mut) {
        dispatch
            .entry("run_id".to_string())
            .or_insert_with(|| Value::String(run_id));
        dispatch
            .entry("task_id".to_string())
            .or_insert_with(|| Value::String(task_id));
        return request;
    }

    if let Some(object) = request.as_object_mut() {
        object
            .entry("run_id".to_string())
            .or_insert_with(|| Value::String(run_id));
        object
            .entry("task_id".to_string())
            .or_insert_with(|| Value::String(task_id));
    }
    request
}

pub(super) fn execute_fan_out_action<E, D>(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    dedupe_key: &str,
    entity_ids: &[String],
    dynamic_artifact: Option<&str>,
    group_by: &[String],
    requires_non_empty: bool,
    max_items: usize,
    fail_fast: bool,
    request_template: &Value,
    executor: E,
    dispatch: &D,
) -> Result<(Value, i32)>
where
    E: AgentTaskExecutorAdapter + Clone,
    D: ControllerDispatchHook,
{
    let expanded_entity_ids = if entity_ids.is_empty() {
        dynamic_artifact
            .map(|artifact| dynamic_fan_out_entity_ids(record, artifact, group_by))
            .unwrap_or_default()
    } else {
        entity_ids.to_vec()
    };

    if expanded_entity_ids.is_empty() {
        let exit_code = if requires_non_empty { 1 } else { 0 };
        return Ok((
            serde_json::json!({ "mode": "fan_out", "item_count": 0, "total_item_count": 0, "max_items": max_items, "fail_fast": fail_fast, "concurrency": 1, "dynamic_artifact": dynamic_artifact, "requires_non_empty": requires_non_empty, "results": [] }),
            exit_code,
        ));
    }

    let mut results = Vec::new();
    let mut exit_code = 0;
    for entity_id in expanded_entity_ids.iter().take(max_items) {
        let request = materialize_fan_out_request(request_template, &record.loop_id, entity_id);
        let child_dedupe_key = format!("{dedupe_key}:{entity_id}");
        let child_action = AgentTaskLoopPolicyActionRecord {
            action_id: format!("{}:{entity_id}", action.action_id),
            action: AgentTaskLoopPolicyAction::SpawnTask {
                dedupe_key: child_dedupe_key.clone(),
                entity_id: Some(entity_id.clone()),
                request: request.clone(),
            },
            status: AgentTaskLoopActionStatus::Running,
            reason: action.reason.clone(),
            created_at: action.created_at.clone(),
            dedupe_key: Some(child_dedupe_key.clone()),
            diagnostics: Vec::new(),
        };
        let (result, child_exit_code) = execute_spawn_task_action(
            record,
            &child_action,
            &child_dedupe_key,
            Some(entity_id),
            &request,
            executor.clone(),
            dispatch,
        )?;
        if child_exit_code != 0 {
            exit_code = child_exit_code;
        }
        results.push(result);
        if child_exit_code != 0 && fail_fast {
            break;
        }
    }
    let item_count = results.len();
    Ok((
        serde_json::json!({ "mode": "fan_out", "item_count": item_count, "total_item_count": expanded_entity_ids.len(), "max_items": max_items, "fail_fast": fail_fast, "concurrency": 1, "dynamic_artifact": dynamic_artifact, "requires_non_empty": requires_non_empty, "truncated": item_count < expanded_entity_ids.len(), "results": results }),
        exit_code,
    ))
}

fn dynamic_fan_out_entity_ids(
    record: &AgentTaskLoopControllerRecord,
    artifact_id: &str,
    group_by: &[String],
) -> Vec<String> {
    let Some(artifact) = find_controller_artifact(record, artifact_id) else {
        return Vec::new();
    };
    let mut items = Vec::new();
    collect_fan_out_items(&artifact, &mut items);
    let mut entity_ids = Vec::new();
    for (index, item) in items.iter().enumerate() {
        let entity_id = fan_out_entity_id(artifact_id, item, group_by)
            .unwrap_or_else(|| format!("{artifact_id}:{}", index + 1));
        if !entity_ids.contains(&entity_id) {
            entity_ids.push(entity_id);
        }
    }
    entity_ids
}

fn collect_fan_out_items(value: &Value, items: &mut Vec<Value>) {
    if let Some(array) = value.as_array() {
        items.extend(array.iter().cloned());
        return;
    }
    let Some(object) = value.as_object() else {
        return;
    };
    for key in ["groups", "items", "records", "artifacts"] {
        if let Some(array) = object.get(key).and_then(Value::as_array) {
            items.extend(array.iter().cloned());
            return;
        }
    }
    if let Some(payload) = object.get("payload") {
        collect_fan_out_items(payload, items);
        return;
    }
    if object.contains_key("group_id") || object.contains_key("id") {
        items.push(value.clone());
    }
}

fn fan_out_entity_id(artifact_id: &str, item: &Value, group_by: &[String]) -> Option<String> {
    if let Some(value) = item.as_str().filter(|value| !value.is_empty()) {
        return Some(value.to_string());
    }
    let object = item.as_object()?;
    for key in ["entity_id", "group_id", "id", "key", "name"] {
        if let Some(value) = object
            .get(key)
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty())
        {
            return Some(format!("{artifact_id}:{value}"));
        }
    }
    if !group_by.is_empty() {
        let values = group_by
            .iter()
            .filter_map(|key| {
                object
                    .get(key)
                    .and_then(Value::as_str)
                    .filter(|value| !value.is_empty())
            })
            .collect::<Vec<_>>();
        if values.len() == group_by.len() {
            return Some(format!("{artifact_id}:{}", values.join(":")));
        }
    }
    None
}

#[allow(clippy::too_many_arguments)]
pub(super) fn execute_spawn_controller_action(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    mode: &str,
    dedupe_key: &str,
    loop_id: &str,
    entity_id: Option<&str>,
    phase: &str,
    config_version: &str,
    request: &Value,
) -> Result<(Value, i32)> {
    let mut created = false;
    let mut child = match controller::load_controller(loop_id) {
        Ok(child) => child,
        Err(_) => {
            created = true;
            let mut child = AgentTaskLoopControllerRecord::new(loop_id, phase, config_version);
            child.parent_loop_id = Some(record.loop_id.clone());
            child.parent_action_id = Some(action.action_id.clone());
            child.parent_entity_id = entity_id.map(str::to_string);
            child.metadata = request.clone();
            child
        }
    };
    let mut child_changed = false;
    if child.parent_loop_id.is_none() {
        child.parent_loop_id = Some(record.loop_id.clone());
        child_changed = true;
    }
    if child.parent_action_id.is_none() {
        child.parent_action_id = Some(action.action_id.clone());
        child_changed = true;
    }
    if child.parent_entity_id.is_none() {
        child.parent_entity_id = entity_id.map(str::to_string);
        child_changed = child.parent_entity_id.is_some();
    }
    if created || child_changed {
        controller::write_controller(&child)?;
    }
    push_controller_history(
        record,
        "controller.action.spawned_controller",
        entity_id.map(str::to_string),
        serde_json::json!({
            "action_id": action.action_id,
            "dedupe_key": dedupe_key,
            "loop_id": child.loop_id,
            "mode": mode,
            "created": created,
        }),
    );
    Ok((
        serde_json::json!({
            "mode": mode,
            "loop_id": child.loop_id,
            "phase": child.phase,
            "config_version": child.config_version,
            "created": created,
        }),
        0,
    ))
}

pub(super) fn execute_wait_for_controller_action(
    record: &mut AgentTaskLoopControllerRecord,
    loop_id: &str,
    entity_id: Option<&str>,
    wait_key: Option<&str>,
    terminal_states: &[AgentTaskLoopControllerState],
) -> Result<(Value, i32)> {
    let wait_key = wait_key
        .map(str::to_string)
        .unwrap_or_else(|| format!("controller:{}:terminal", loop_id.replace(['/', ':'], "_")));
    let terminal_states = if terminal_states.is_empty() {
        vec![
            AgentTaskLoopControllerState::Completed,
            AgentTaskLoopControllerState::Failed,
            AgentTaskLoopControllerState::HumanReady,
            AgentTaskLoopControllerState::Abandoned,
            AgentTaskLoopControllerState::Escalated,
        ]
    } else {
        terminal_states.to_vec()
    };
    let child_state = controller::controller_status(loop_id)
        .ok()
        .map(|child| child.state);
    let satisfied = child_state.is_some_and(|state| terminal_states.contains(&state));
    if satisfied {
        for wait in &mut record.waits {
            if wait.wait_key == wait_key && wait.status == controller::AgentTaskLoopWaitStatus::Open
            {
                wait.status = controller::AgentTaskLoopWaitStatus::Satisfied;
                wait.satisfied_by_event_id =
                    child_state.map(|state| format!("controller-terminal:{loop_id}:{state:?}"));
            }
        }
        if record.state == AgentTaskLoopControllerState::Waiting {
            record.state = AgentTaskLoopControllerState::Running;
        }
    } else {
        record.state = AgentTaskLoopControllerState::Waiting;
    }
    Ok((
        serde_json::json!({
            "mode": "wait_for_controller",
            "loop_id": loop_id,
            "entity_id": entity_id,
            "wait_key": wait_key,
            "terminal_states": terminal_states,
            "child_state": child_state,
            "satisfied": satisfied,
        }),
        0,
    ))
}

/// A gate bundle blocks the loop (non-zero exit) when it failed outright or is
/// still pending an external/manual result. Passed and non-blocking Warn
/// bundles allow the loop to advance.
fn gate_bundle_exit_code(status: AgentTaskGateBundleStatus) -> i32 {
    match status {
        AgentTaskGateBundleStatus::Failed | AgentTaskGateBundleStatus::Pending => 1,
        AgentTaskGateBundleStatus::Passed | AgentTaskGateBundleStatus::Warn => 0,
    }
}

pub(super) fn execute_run_gates_action(
    record: &mut AgentTaskLoopControllerRecord,
    bundle_id: &str,
    entity_id: Option<&str>,
) -> Result<(Value, i32)> {
    if let Some(existing) = record
        .gate_results
        .iter()
        .rev()
        .find(|result| result.bundle_id == bundle_id && result.entity_id.as_deref() == entity_id)
        .cloned()
    {
        let exit_code = gate_bundle_exit_code(existing.status);
        return Ok((
            serde_json::json!({ "mode": "run_gates", "bundle_id": bundle_id, "entity_id": entity_id, "result": existing }),
            exit_code,
        ));
    }

    let bundle = record
        .gate_bundles
        .iter()
        .find(|bundle| bundle.bundle_id == bundle_id)
        .cloned()
        .ok_or_else(|| {
            Error::validation_invalid_argument(
                "bundle_id",
                format!("gate bundle '{bundle_id}' does not exist"),
                Some(bundle_id.to_string()),
                None,
            )
        })?;

    let mut checks = Vec::new();
    for check in &bundle.checks {
        let result = match check.kind {
            AgentTaskGateBundleCheckKind::Command => run_command_gate_check(check)?,
            AgentTaskGateBundleCheckKind::Manual => AgentTaskGateCheckResult {
                check_id: check.check_id.clone(),
                // A manual check requires an external result that has not
                // arrived. Treat it as Pending (blocking) rather than a
                // non-blocking Warn so a manual-only bundle never resolves to a
                // false-green acceptance gate.
                status: AgentTaskGateBundleStatus::Pending,
                retryable: check.retryable,
                classification: Some("manual_gate_requires_external_result".to_string()),
                evidence: Vec::new(),
                details: serde_json::json!({ "input": check.input }),
            },
            AgentTaskGateBundleCheckKind::Api | AgentTaskGateBundleCheckKind::Tool => {
                AgentTaskGateCheckResult {
                    check_id: check.check_id.clone(),
                    status: AgentTaskGateBundleStatus::Failed,
                    retryable: check.retryable,
                    classification: Some("unsupported_generic_gate_kind".to_string()),
                    evidence: Vec::new(),
                    details: serde_json::json!({ "kind": check.kind, "input": check.input }),
                }
            }
        };
        checks.push(result);
    }

    let status = if checks
        .iter()
        .any(|check| check.status == AgentTaskGateBundleStatus::Failed)
    {
        AgentTaskGateBundleStatus::Failed
    } else if checks
        .iter()
        .any(|check| check.status == AgentTaskGateBundleStatus::Pending)
    {
        AgentTaskGateBundleStatus::Pending
    } else if checks
        .iter()
        .any(|check| check.status == AgentTaskGateBundleStatus::Warn)
    {
        AgentTaskGateBundleStatus::Warn
    } else {
        AgentTaskGateBundleStatus::Passed
    };
    let result = AgentTaskGateBundleResult {
        result_id: format!("gate-result-{}", record.gate_results.len() + 1),
        bundle_id: bundle_id.to_string(),
        entity_id: entity_id.map(str::to_string),
        run_id: None,
        status,
        checks,
        recorded_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
    };
    record.gate_results.push(result.clone());
    let exit_code = gate_bundle_exit_code(status);
    Ok((
        serde_json::json!({ "mode": "run_gates", "bundle_id": bundle_id, "entity_id": entity_id, "result": result }),
        exit_code,
    ))
}

pub(super) fn run_command_gate_check(
    check: &crate::core::agent_task_loop_controller::AgentTaskGateBundleCheck,
) -> Result<AgentTaskGateCheckResult> {
    let command = check
        .input
        .get("command")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            Error::validation_invalid_argument(
                "command",
                "command gate check requires input.command",
                Some(check.input.to_string()),
                None,
            )
        })?;
    let cwd = check.input.get("cwd").and_then(Value::as_str);
    let timeout_seconds = check
        .input
        .get("timeout_seconds")
        .and_then(Value::as_u64)
        .filter(|seconds| *seconds > 0)
        .unwrap_or(COMMAND_GATE_DEFAULT_TIMEOUT_SECONDS);

    let mut child_command = Command::new("sh");
    child_command
        .arg("-lc")
        .arg(command)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    if let Some(cwd) = cwd {
        child_command.current_dir(cwd);
    }

    let mut child = child_command.spawn().map_err(|error| {
        Error::internal_io(
            format!("failed to execute gate command '{command}': {error}"),
            None,
        )
    })?;
    let stdout = child.stdout.take().map(read_capped_command_output);
    let stderr = child.stderr.take().map(read_capped_command_output);
    let deadline = Instant::now() + Duration::from_secs(timeout_seconds);
    let mut timed_out = false;
    let exit_status = loop {
        if let Some(status) = child.try_wait().map_err(|error| {
            Error::internal_io(
                format!("failed to poll gate command '{command}': {error}"),
                None,
            )
        })? {
            break status;
        }
        if Instant::now() >= deadline {
            timed_out = true;
            let _ = child.kill();
            break child.wait().map_err(|error| {
                Error::internal_io(
                    format!("failed to reap timed out gate command '{command}': {error}"),
                    None,
                )
            })?;
        }
        thread::sleep(Duration::from_millis(25));
    };
    let stdout = collect_capped_command_output(stdout, command, "stdout")?;
    let stderr = collect_capped_command_output(stderr, command, "stderr")?;

    let status = if !timed_out && exit_status.success() {
        AgentTaskGateBundleStatus::Passed
    } else {
        AgentTaskGateBundleStatus::Failed
    };
    Ok(AgentTaskGateCheckResult {
        check_id: check.check_id.clone(),
        status,
        retryable: check.retryable,
        classification: timed_out.then(|| "command_timed_out".to_string()),
        evidence: Vec::new(),
        details: serde_json::json!({
            "command": command,
            "cwd": cwd,
            "timeout_seconds": timeout_seconds,
            "timed_out": timed_out,
            "exit_code": exit_status.code(),
            "stdout": String::from_utf8_lossy(&stdout.bytes).into_owned(),
            "stderr": String::from_utf8_lossy(&stderr.bytes).into_owned(),
            "stdout_bytes": stdout.total_bytes,
            "stderr_bytes": stderr.total_bytes,
            "stdout_stored_bytes": stdout.bytes.len(),
            "stderr_stored_bytes": stderr.bytes.len(),
            "stdout_truncated": stdout.truncated,
            "stderr_truncated": stderr.truncated,
        }),
    })
}

struct CappedCommandOutput {
    bytes: Vec<u8>,
    total_bytes: usize,
    truncated: bool,
}

fn read_capped_command_output<R>(
    mut reader: R,
) -> thread::JoinHandle<io::Result<CappedCommandOutput>>
where
    R: Read + Send + 'static,
{
    thread::spawn(move || {
        let mut bytes = Vec::with_capacity(COMMAND_GATE_OUTPUT_CAP_BYTES);
        let mut total_bytes = 0usize;
        let mut buffer = [0u8; 8192];
        loop {
            let read = reader.read(&mut buffer)?;
            if read == 0 {
                break;
            }
            total_bytes = total_bytes.saturating_add(read);
            if bytes.len() < COMMAND_GATE_OUTPUT_CAP_BYTES {
                let remaining = COMMAND_GATE_OUTPUT_CAP_BYTES - bytes.len();
                bytes.extend_from_slice(&buffer[..read.min(remaining)]);
            }
        }
        Ok(CappedCommandOutput {
            bytes,
            total_bytes,
            truncated: total_bytes > COMMAND_GATE_OUTPUT_CAP_BYTES,
        })
    })
}

fn collect_capped_command_output(
    handle: Option<thread::JoinHandle<io::Result<CappedCommandOutput>>>,
    command: &str,
    stream: &str,
) -> Result<CappedCommandOutput> {
    handle
        .ok_or_else(|| {
            Error::internal_io(format!("gate command '{command}' missing {stream}"), None)
        })?
        .join()
        .map_err(|_| {
            Error::internal_io(
                format!("gate command '{command}' {stream} reader panicked"),
                None,
            )
        })?
        .map_err(|error| {
            Error::internal_io(
                format!("failed to read gate command '{command}' {stream}: {error}"),
                None,
            )
        })
}

#[allow(clippy::too_many_arguments)]
pub(super) fn execute_route_finding_action<E, D>(
    record: &mut AgentTaskLoopControllerRecord,
    action: &AgentTaskLoopPolicyActionRecord,
    finding: &controller::AgentTaskLoopFindingPacket,
    dedupe_key: &str,
    entity_id: Option<&str>,
    request_template: &Value,
    executor: E,
    dispatch: &D,
) -> Result<(Value, i32)>
where
    E: AgentTaskExecutorAdapter + Clone,
    D: ControllerDispatchHook,
{
    let entity_id = ensure_finding_entity(record, finding, entity_id);
    let request = materialize_route_finding_request(request_template, finding, &entity_id);
    execute_spawn_task_action(
        record,
        action,
        dedupe_key,
        Some(&entity_id),
        &request,
        executor,
        dispatch,
    )
}

pub(super) fn ensure_finding_entity(
    record: &mut AgentTaskLoopControllerRecord,
    finding: &controller::AgentTaskLoopFindingPacket,
    entity_id: Option<&str>,
) -> String {
    if let Some(entity_id) = entity_id {
        return entity_id.to_string();
    }
    let key = finding
        .reproduction_key
        .as_deref()
        .unwrap_or(&finding.finding_id);
    let entity_id = format!("finding:{}", key.replace(['/', ':', '#', ' '], "_"));
    record
        .entities
        .entry(entity_id.clone())
        .or_insert_with(|| AgentTaskLoopEntity {
            entity_id: entity_id.clone(),
            entity_type: "finding".to_string(),
            key: key.to_string(),
            dedupe_key: format!("entity:finding:{key}"),
            state: Some("routed".to_string()),
            human_ready: false,
            parent_entity_ids: Vec::new(),
            run_refs: Vec::new(),
            artifact_refs: finding.lineage.clone(),
            provenance: finding
                .lineage
                .iter()
                .map(|artifact| AgentTaskLoopProvenanceRef {
                    kind: artifact
                        .kind
                        .clone()
                        .unwrap_or_else(|| "artifact".to_string()),
                    uri: artifact.uri.clone(),
                    caused_by: Some(finding.finding_id.clone()),
                })
                .collect(),
            metadata: serde_json::json!({
                "severity": finding.severity,
                "owner": finding.owner,
                "source_transformer": finding.source_transformer,
            }),
        });
    entity_id
}

pub(super) fn materialize_route_finding_request(
    template: &Value,
    finding: &controller::AgentTaskLoopFindingPacket,
    entity_id: &str,
) -> Value {
    let mut request = template.clone();
    if let Some(object) = request.as_object_mut() {
        object.insert(
            "entity_id".to_string(),
            Value::String(entity_id.to_string()),
        );
        object.insert(
            "finding_id".to_string(),
            Value::String(finding.finding_id.clone()),
        );
        object
            .entry("finding".to_string())
            .or_insert_with(|| serde_json::to_value(finding).unwrap_or(Value::Null));
    }
    request
}
