use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use std::time::{Duration, Instant};

use reqwest::blocking::Client;
use serde::Serialize;
use serde_json::json;

use crate::core::api_jobs::{
    Job, JobArtifactMetadata, JobStatus, RemoteRunnerJobClaim, RemoteRunnerJobRequest,
    RemoteRunnerJobResult,
};
use crate::core::error::{Error, Result};

use super::broker_http;
use super::capabilities::RunnerCapabilityPreflight;
use super::execution::{exec_worker_local_until_cancelled, RunnerExecOptions};

const BROKER_CANCEL_POLL_INTERVAL: Duration = Duration::from_millis(250);

#[derive(Debug, Clone)]
pub struct ReverseRunnerWorkerOptions {
    pub runner_id: String,
    pub broker_url: String,
    /// Paired broker bearer token. Required when the broker enforces auth;
    /// omitted for loopback-open smoke setups.
    pub broker_token: Option<String>,
    pub project_id: Option<String>,
    pub lease_ms: u64,
    pub concurrency_limit: Option<usize>,
    pub loop_mode: bool,
    pub idle_backoff_ms: u64,
    pub max_idle_backoff_ms: u64,
    pub broker_failure_backoff_ms: u64,
    pub broker_retry_limit: u32,
}

#[derive(Debug, Clone, Serialize)]
pub struct ReverseRunnerWorkerOutput {
    pub variant: &'static str,
    pub command: &'static str,
    pub runner_id: String,
    pub broker_url: String,
    pub claimed: bool,
    #[serde(skip_serializing_if = "is_false")]
    pub loop_mode: bool,
    #[serde(skip_serializing_if = "is_zero")]
    pub iterations: u64,
    #[serde(skip_serializing_if = "is_zero")]
    pub jobs_claimed: u64,
    #[serde(skip_serializing_if = "is_zero")]
    pub broker_failures: u32,
    #[serde(skip_serializing_if = "is_false")]
    pub stopped: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_claim: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_result: Option<i32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_error: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub job: Option<Job>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub exit_code: Option<i32>,
}

fn is_false(value: &bool) -> bool {
    !*value
}

fn is_zero<T>(value: &T) -> bool
where
    T: PartialEq + From<u8>,
{
    *value == T::from(0)
}

pub fn run_reverse_worker(
    options: ReverseRunnerWorkerOptions,
) -> Result<(ReverseRunnerWorkerOutput, i32)> {
    let stop = Arc::new(AtomicBool::new(false));
    if options.loop_mode {
        crate::core::process::install_shutdown_handler(stop.clone(), "runner worker")?;
    }
    run_reverse_worker_with_stop(options, stop)
}

fn run_reverse_worker_with_stop(
    options: ReverseRunnerWorkerOptions,
    stop: Arc<AtomicBool>,
) -> Result<(ReverseRunnerWorkerOutput, i32)> {
    if options.runner_id.trim().is_empty() {
        return Err(Error::validation_invalid_argument(
            "runner_id",
            "reverse runner worker requires a runner id",
            None,
            None,
        ));
    }
    if options.loop_mode {
        run_loop(options, stop, std::thread::sleep)
    } else {
        run_once_output(options, 1, 0, 0, false)
    }
}

fn run_loop(
    options: ReverseRunnerWorkerOptions,
    stop: Arc<AtomicBool>,
    mut sleep: impl FnMut(Duration),
) -> Result<(ReverseRunnerWorkerOutput, i32)> {
    let mut iterations = 0;
    let mut jobs_claimed = 0;
    let mut broker_failures = 0;
    let mut idle_backoff = Duration::from_millis(options.idle_backoff_ms.max(1));
    let max_idle_backoff = Duration::from_millis(options.max_idle_backoff_ms.max(1));
    let broker_failure_backoff = Duration::from_millis(options.broker_failure_backoff_ms.max(1));
    let mut last_job = None;
    let mut last_exit_code = None;
    let mut last_claim = None;
    let mut last_result = None;
    let mut last_error = None;

    log_worker_event(
        &options,
        "started",
        json!({
            "loop": true,
            "idle_backoff_ms": idle_backoff.as_millis(),
            "max_idle_backoff_ms": max_idle_backoff.as_millis(),
            "broker_failure_backoff_ms": broker_failure_backoff.as_millis(),
            "broker_retry_limit": options.broker_retry_limit,
        }),
    );

    while !stop.load(Ordering::SeqCst) {
        iterations += 1;
        match run_once_output(
            options.clone(),
            iterations,
            jobs_claimed,
            broker_failures,
            false,
        ) {
            Ok((output, exit_code)) => {
                broker_failures = 0;
                if output.claimed {
                    jobs_claimed += 1;
                    idle_backoff = Duration::from_millis(options.idle_backoff_ms.max(1));
                    last_claim = output.job.as_ref().map(|job| job.id.to_string());
                    last_result = Some(exit_code);
                    if exit_code == 0 {
                        last_error = None;
                    } else {
                        last_error = Some(format!("job exited with code {exit_code}"));
                    }
                    last_job = output.job;
                    last_exit_code = Some(exit_code);
                    log_worker_event(
                        &options,
                        "job_finished",
                        json!({
                            "iteration": iterations,
                            "job_id": last_claim.as_ref(),
                            "exit_code": exit_code,
                            "jobs_claimed": jobs_claimed,
                        }),
                    );
                } else {
                    log_worker_event(
                        &options,
                        "idle",
                        json!({
                            "iteration": iterations,
                            "sleep_ms": idle_backoff.as_millis(),
                        }),
                    );
                    sleep(idle_backoff);
                    idle_backoff = std::cmp::min(idle_backoff.saturating_mul(2), max_idle_backoff);
                }
            }
            Err(err) => {
                broker_failures = broker_failures.saturating_add(1);
                last_error = Some(err.to_string());
                log_worker_event(
                    &options,
                    "broker_failure",
                    json!({
                        "iteration": iterations,
                        "broker_failures": broker_failures,
                        "retry_limit": options.broker_retry_limit,
                        "sleep_ms": broker_failure_backoff.as_millis(),
                        "error": err.to_string(),
                    }),
                );
                if broker_failures > options.broker_retry_limit {
                    return Err(err);
                }
                sleep(broker_failure_backoff);
            }
        }
    }

    log_worker_event(
        &options,
        "stopped",
        json!({
            "iterations": iterations,
            "jobs_claimed": jobs_claimed,
            "broker_failures": broker_failures,
        }),
    );
    Ok((
        ReverseRunnerWorkerOutput {
            variant: "work",
            command: "runner.work",
            runner_id: options.runner_id,
            broker_url: options.broker_url,
            claimed: jobs_claimed > 0,
            loop_mode: true,
            iterations,
            jobs_claimed,
            broker_failures,
            stopped: true,
            last_claim,
            last_result,
            last_error,
            job: last_job,
            exit_code: last_exit_code,
        },
        0,
    ))
}

fn run_once_output(
    options: ReverseRunnerWorkerOptions,
    iterations: u64,
    jobs_claimed: u64,
    broker_failures: u32,
    stopped: bool,
) -> Result<(ReverseRunnerWorkerOutput, i32)> {
    if options.broker_url.trim().is_empty() {
        return Err(Error::validation_invalid_argument(
            "broker_url",
            "reverse runner worker requires a broker URL",
            None,
            None,
        ));
    }

    let client = Client::builder()
        .timeout(Duration::from_secs(10))
        .build()
        .map_err(|err| Error::internal_unexpected(format!("build broker HTTP client: {err}")))?;
    let claim = claim_job(&client, &options)?;
    let Some(claim) = claim else {
        let loop_mode = options.loop_mode;
        return Ok((
            ReverseRunnerWorkerOutput {
                variant: "work",
                command: "runner.work",
                runner_id: options.runner_id,
                broker_url: options.broker_url,
                claimed: false,
                loop_mode,
                iterations: if loop_mode { iterations } else { 0 },
                jobs_claimed: if loop_mode { jobs_claimed } else { 0 },
                broker_failures: if loop_mode { broker_failures } else { 0 },
                stopped,
                last_claim: None,
                last_result: None,
                last_error: None,
                job: None,
                exit_code: None,
            },
            0,
        ));
    };

    if let Some(job) = cancelled_job_snapshot(
        &client,
        &options.broker_url,
        options.broker_token.as_deref(),
        &claim.job,
    )? {
        return Ok(cancelled_output(
            options,
            iterations,
            jobs_claimed,
            broker_failures,
            stopped,
            job,
        ));
    }

    append_progress(
        &client,
        &options.broker_url,
        options.broker_token.as_deref(),
        &options.runner_id,
        &claim.job,
    )?;
    // Remote capability-parity preflight: validate that this runner can satisfy
    // the claimed job's top-level command (and any required paths) before
    // starting execution, so a missing tool fails before remote dispatch instead
    // of mid-run (#5093). The local worker execution path runs this preflight
    // before handing the claimed job directly to the local runtime.
    let capability_preflight = reverse_worker_capability_preflight(&claim.request);
    // Shared finisher so the exec-error and exec-success paths submit their
    // terminal job result through identical broker plumbing (#5091).
    let finish = |result: RemoteRunnerJobResult| {
        finish_job(
            &client,
            &options.broker_url,
            options.broker_token.as_deref(),
            &options.runner_id,
            &claim.job,
            result,
        )
    };
    let mut cancel_seen = false;
    let mut last_cancel_poll = Instant::now();
    let exec_result = exec_worker_local_until_cancelled(
        &options.runner_id,
        RunnerExecOptions {
            cwd: claim.request.cwd.clone(),
            project_id: claim.request.project_id.clone(),
            allow_diagnostic_ssh: false,
            command: claim.request.command.clone(),
            env: claim.request.env.clone(),
            secret_env_names: claim.request.secret_env_names.clone(),
            capture_patch: claim.request.capture_patch,
            raw_exec: false,
            source_snapshot: claim.request.source_snapshot.clone(),
            capability_preflight,
            required_extensions: Vec::new(),
            require_paths: claim.request.require_paths.clone(),
            detach_after_handoff: false,
        },
        || {
            if cancel_seen || last_cancel_poll.elapsed() < BROKER_CANCEL_POLL_INTERVAL {
                return cancel_seen;
            }
            last_cancel_poll = Instant::now();
            cancel_seen = cancelled_job_snapshot(
                &client,
                &options.broker_url,
                options.broker_token.as_deref(),
                &claim.job,
            )
            .map(|job| job.is_some())
            .unwrap_or(false);
            cancel_seen
        },
    );
    let (exec_output, exit_code) = match exec_result {
        Ok(result) => result,
        Err(err) => {
            if let Some(job) = cancelled_job_snapshot(
                &client,
                &options.broker_url,
                options.broker_token.as_deref(),
                &claim.job,
            )? {
                return Ok(cancelled_output(
                    options,
                    iterations,
                    jobs_claimed,
                    broker_failures,
                    stopped,
                    job,
                ));
            }
            let job = finish(RemoteRunnerJobResult {
                exit_code: 1,
                stdout: None,
                stderr: Some(err.to_string()),
                patch: None,
                mutation_artifacts: None,
                data: Some(json!({
                    "error": err.to_string(),
                })),
                observation_run_ids: Vec::new(),
                artifacts: Vec::new(),
                artifact_refs: Vec::new(),
                metrics: None,
                capture: None,
            })?;
            let exit_code = 1;
            return Ok((
                claimed_output(
                    options,
                    iterations,
                    jobs_claimed,
                    broker_failures,
                    stopped,
                    job,
                    exit_code,
                ),
                exit_code,
            ));
        }
    };
    if let Some(job) = cancelled_job_snapshot(
        &client,
        &options.broker_url,
        options.broker_token.as_deref(),
        &claim.job,
    )? {
        return Ok(cancelled_output(
            options,
            iterations,
            jobs_claimed,
            broker_failures,
            stopped,
            job,
        ));
    }
    let job = finish(remote_runner_result_from_exec_output(
        exec_output,
        exit_code,
    ))?;

    Ok((
        claimed_output(
            options,
            iterations,
            jobs_claimed,
            broker_failures,
            stopped,
            job,
            exit_code,
        ),
        exit_code,
    ))
}

fn remote_runner_result_from_exec_output(
    exec_output: super::execution::RunnerExecOutput,
    exit_code: i32,
) -> RemoteRunnerJobResult {
    let patch = exec_output.patch.clone();
    let mutation_artifacts = exec_output.mutation_artifacts.clone();
    let mut data = json!({
        "mode": exec_output.mode,
        "remote_cwd": exec_output.remote_cwd,
    });
    if let Some(patch) = patch.clone() {
        data["patch"] = patch;
    }
    if let Some(mutation_artifacts) = mutation_artifacts.clone() {
        data["mutation_artifacts"] =
            serde_json::to_value(&mutation_artifacts).unwrap_or(serde_json::Value::Null);
    }
    if let Some(mirror_run_id) = exec_output.mirror_run_id.clone() {
        data["mirror_run_id"] = json!(mirror_run_id);
    }
    RemoteRunnerJobResult {
        exit_code,
        stdout: Some(exec_output.stdout),
        stderr: Some(exec_output.stderr),
        patch,
        mutation_artifacts,
        data: Some(data),
        observation_run_ids: exec_output.mirror_run_id.into_iter().collect(),
        artifacts: exec_output.artifacts,
        artifact_refs: exec_output
            .runner_result
            .map(|result| {
                result
                    .artifact_refs
                    .into_iter()
                    .map(|artifact| JobArtifactMetadata {
                        id: artifact.artifact_id,
                        name: artifact.name,
                        path: artifact.path,
                        url: artifact.url,
                        mime: artifact.mime,
                        size_bytes: artifact.size_bytes,
                        sha256: artifact.sha256,
                        metadata: None,
                    })
                    .collect()
            })
            .unwrap_or_default(),
        metrics: exec_output.metrics,
        capture: exec_output.capture,
    }
}

fn cancelled_output(
    options: ReverseRunnerWorkerOptions,
    iterations: u64,
    jobs_claimed: u64,
    broker_failures: u32,
    stopped: bool,
    job: Job,
) -> (ReverseRunnerWorkerOutput, i32) {
    let exit_code = 0;
    (
        claimed_output(
            options,
            iterations,
            jobs_claimed,
            broker_failures,
            stopped,
            job,
            exit_code,
        ),
        exit_code,
    )
}

/// Build the remote capability-parity preflight for a claimed reverse-runner
/// job. The claimed command's executable (its first argv element) must be
/// available on this runner before execution starts, mirroring the direct
/// `runner exec` path's preflight contract (#5093).
fn reverse_worker_capability_preflight(
    request: &RemoteRunnerJobRequest,
) -> Option<RunnerCapabilityPreflight> {
    let required_commands: Vec<String> = request
        .command
        .first()
        .filter(|program| !program.trim().is_empty())
        .cloned()
        .into_iter()
        .collect();
    if required_commands.is_empty() {
        return None;
    }
    Some(RunnerCapabilityPreflight {
        command: "runner.work".to_string(),
        required_commands,
        ..Default::default()
    })
}

fn claimed_output(
    options: ReverseRunnerWorkerOptions,
    iterations: u64,
    jobs_claimed: u64,
    broker_failures: u32,
    stopped: bool,
    job: Job,
    exit_code: i32,
) -> ReverseRunnerWorkerOutput {
    let loop_mode = options.loop_mode;
    ReverseRunnerWorkerOutput {
        variant: "work",
        command: "runner.work",
        runner_id: options.runner_id,
        broker_url: options.broker_url,
        claimed: true,
        loop_mode,
        iterations: if loop_mode { iterations } else { 0 },
        jobs_claimed: if loop_mode { jobs_claimed + 1 } else { 0 },
        broker_failures: if loop_mode { broker_failures } else { 0 },
        stopped,
        last_claim: if loop_mode {
            Some(job.id.to_string())
        } else {
            None
        },
        last_result: if loop_mode { Some(exit_code) } else { None },
        last_error: if loop_mode && exit_code != 0 {
            Some(format!("job exited with code {exit_code}"))
        } else {
            None
        },
        job: Some(job),
        exit_code: Some(exit_code),
    }
}

fn log_worker_event(options: &ReverseRunnerWorkerOptions, event: &str, data: serde_json::Value) {
    eprintln!(
        "{}",
        json!({
            "command": "runner.work",
            "event": event,
            "runner_id": options.runner_id,
            "broker_url": options.broker_url,
            "project_id": options.project_id,
            "data": data,
        })
    );
}

fn claim_job(
    client: &Client,
    options: &ReverseRunnerWorkerOptions,
) -> Result<Option<RemoteRunnerJobClaim>> {
    let data = broker_http::post_json(
        client,
        &options.broker_url,
        "/runner/jobs/claim",
        json!({
            "runner_id": options.runner_id,
            "project_id": options.project_id,
            "lease_ms": options.lease_ms.max(1),
            "concurrency_limit": options.concurrency_limit,
        }),
        "claim reverse runner job",
        options.broker_token.as_deref(),
    )?;
    let claim = data["claim"].clone();
    if claim.is_null() {
        return Ok(None);
    }
    serde_json::from_value(claim).map(Some).map_err(|err| {
        Error::internal_json(
            err.to_string(),
            Some("parse reverse runner job claim".to_string()),
        )
    })
}

fn append_progress(
    client: &Client,
    broker_url: &str,
    token: Option<&str>,
    runner_id: &str,
    job: &Job,
) -> Result<()> {
    let claim_id = remote_runner_claim_id(job)?;
    broker_http::post_json(
        client,
        broker_url,
        &format!("/runner/jobs/{}/events", job.id),
        json!({
            "runner_id": runner_id,
            "claim_id": claim_id,
            "kind": "progress",
            "message": "reverse runner worker started execution",
        }),
        "append reverse runner progress event",
        token,
    )?;
    Ok(())
}

fn finish_job(
    client: &Client,
    broker_url: &str,
    token: Option<&str>,
    runner_id: &str,
    job: &Job,
    result: RemoteRunnerJobResult,
) -> Result<Job> {
    let claim_id = remote_runner_claim_id(job)?;
    let data = broker_http::post_json(
        client,
        broker_url,
        &format!("/runner/jobs/{}/finish", job.id),
        json!({
            "runner_id": runner_id,
            "claim_id": claim_id,
            "result": result,
        }),
        "finish reverse runner job",
        token,
    )?;
    serde_json::from_value(data["job"].clone()).map_err(|err| {
        Error::internal_json(
            err.to_string(),
            Some("parse finished reverse runner job".to_string()),
        )
    })
}

fn cancelled_job_snapshot(
    client: &Client,
    broker_url: &str,
    token: Option<&str>,
    job: &Job,
) -> Result<Option<Job>> {
    let data = broker_http::get_json(
        client,
        broker_url,
        &format!("/jobs/{}", job.id),
        "inspect reverse runner job cancellation state",
        token,
    )?;
    let snapshot: Job = serde_json::from_value(data["job"].clone()).map_err(|err| {
        Error::internal_json(
            err.to_string(),
            Some("parse reverse runner job snapshot".to_string()),
        )
    })?;
    if snapshot.status == JobStatus::Cancelled {
        Ok(Some(snapshot))
    } else {
        Ok(None)
    }
}

fn remote_runner_claim_id(job: &Job) -> Result<&str> {
    job.claim_id.as_deref().ok_or_else(|| {
        Error::validation_invalid_argument(
            "claim_id",
            "claimed remote runner job is missing a claim id",
            Some(job.id.to_string()),
            None,
        )
    })
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Write};

    use super::*;
    use crate::core::api_jobs::{
        JobArtifactMetadata, JobEventKind, JobStatus, JobStore, RemoteRunnerJobRequest,
    };
    use crate::core::runner::RunnerExecMode;
    use crate::core::server::RunnerPolicy;
    use crate::test_support;
    use serde_json::Value;

    fn worker_options(broker_url: String) -> ReverseRunnerWorkerOptions {
        ReverseRunnerWorkerOptions {
            runner_id: "lab".to_string(),
            broker_url,
            broker_token: None,
            project_id: None,
            lease_ms: 30_000,
            concurrency_limit: None,
            loop_mode: false,
            idle_backoff_ms: 1,
            max_idle_backoff_ms: 10,
            broker_failure_backoff_ms: 1,
            broker_retry_limit: 1,
        }
    }

    #[test]
    fn reverse_worker_result_preserves_exec_patch_and_artifacts() {
        let result = remote_runner_result_from_exec_output(
            super::super::execution::RunnerExecOutput {
                variant: "exec",
                command: "runner.exec",
                runner_id: "lab".to_string(),
                dry_run: false,
                mode: RunnerExecMode::Local,
                argv: vec!["homeboy".to_string(), "refactor".to_string()],
                remote_cwd: "/srv/workspace".to_string(),
                exit_code: 0,
                stdout: "ok".to_string(),
                stderr: String::new(),
                source_snapshot: None,
                job: None,
                runner_job: None,
                job_id: None,
                job_events: None,
                mirror_run_id: None,
                patch: Some(json!({
                    "patch_artifact_id": "patch.diff",
                    "modified_files": ["src/lib.rs"],
                })),
                mutation_artifacts: Some(crate::core::runner::RunnerMutationArtifacts {
                    patch_ref: Some(crate::core::runner::RunnerArtifactRef {
                        artifact_id: "patch.diff".to_string(),
                        name: Some("patch.diff".to_string()),
                        path: Some("/srv/workspace/.homeboy/patch.diff".to_string()),
                        url: None,
                        mime: Some("text/x-diff".to_string()),
                        size_bytes: Some(42),
                        sha256: Some("abc123".to_string()),
                        transport: None,
                    }),
                    file_bundle_ref: None,
                    operation_log_ref: None,
                }),
                artifacts: vec![JobArtifactMetadata {
                    id: "patch.diff".to_string(),
                    name: Some("patch.diff".to_string()),
                    path: Some("/srv/workspace/.homeboy/patch.diff".to_string()),
                    url: None,
                    mime: Some("text/x-diff".to_string()),
                    size_bytes: Some(42),
                    sha256: Some("abc123".to_string()),
                    metadata: Some(json!({ "kind": "lab_fix_patch" })),
                }],
                metrics: None,
                capture: None,
                runner_result: None,
                handoff: None,
                diagnostics: None,
            },
            0,
        );

        assert_eq!(result.exit_code, 0);
        assert_eq!(
            result.patch.as_ref().expect("patch")["patch_artifact_id"],
            "patch.diff"
        );
        assert_eq!(
            result.data.as_ref().expect("data")["patch"]["modified_files"][0],
            "src/lib.rs"
        );
        assert_eq!(result.artifacts[0].id, "patch.diff");
        assert_eq!(
            result
                .mutation_artifacts
                .as_ref()
                .and_then(|artifacts| artifacts.patch_ref.as_ref())
                .map(|artifact| artifact.artifact_id.as_str()),
            Some("patch.diff")
        );
    }

    #[test]
    fn reverse_worker_executes_claimed_job_and_finishes_it() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            crate::core::runner::merge(
                Some("lab"),
                &serde_json::json!({
                    "policy": RunnerPolicy {
                        allow_raw_exec: Some(true),
                        workspace_roots: vec!["/tmp".to_string()],
                        allowed_commands: vec!["sh".to_string()],
                        ..Default::default()
                    }
                })
                .to_string(),
                &[],
            )
            .expect("set policy");
            let store = JobStore::default();
            store
                .submit_remote_runner_job(RemoteRunnerJobRequest {
                    runner_id: "lab".to_string(),
                    project_id: None,
                    operation: "runner.exec".to_string(),
                    command: vec![
                        "sh".to_string(),
                        "-c".to_string(),
                        "printf worker-ok".to_string(),
                    ],
                    cwd: Some("/tmp".to_string()),
                    env: Default::default(),
                    secret_env_names: Vec::new(),
                    capture_patch: false,
                    source_snapshot: None,
                    require_paths: Vec::new(),
                    metadata: None,
                })
                .expect("submit job");
            let seen_paths = Arc::new(std::sync::Mutex::new(Vec::new()));
            let (broker_url, handle) =
                spawn_mock_broker_with_paths(store.clone(), 5, Some(seen_paths.clone()));
            write_reverse_controller_session(&broker_url);

            let (output, exit_code) =
                run_reverse_worker(worker_options(broker_url.clone())).expect("run worker");

            assert!(output.claimed);
            let serialized = serde_json::to_value(&output).expect("serialize output");
            assert_eq!(serialized["command"], serde_json::json!("runner.work"));
            assert_eq!(serialized["claimed"], serde_json::json!(true));
            assert!(serialized.get("loop_mode").is_none());
            assert!(serialized.get("iterations").is_none());
            assert!(serialized.get("jobs_claimed").is_none());
            assert!(serialized.get("last_claim").is_none());
            let job = output.job.clone().expect("job");
            let events = store.events(job.id).expect("events");
            let result = events
                .iter()
                .find(|event| event.kind == JobEventKind::Result)
                .and_then(|event| event.data.as_ref())
                .expect("result event data");
            assert_eq!(
                exit_code, 0,
                "worker output: {output:#?}; result: {result:#}"
            );
            assert_eq!(job.status, JobStatus::Succeeded);
            handle.join().expect("mock broker joins");
            assert!(events.iter().any(|event| {
                event.kind == JobEventKind::Result
                    && event.data.as_ref().expect("result data")["stdout"]
                        == serde_json::json!("worker-ok")
            }));
            assert!(result["metrics"]["duration_ms"].as_u64().is_some());
            let seen_paths = seen_paths.lock().expect("seen paths");
            assert!(
                !seen_paths.iter().any(|path| path == "/runner/jobs"),
                "claimed worker job must execute locally instead of submitting another reverse broker job"
            );
            if cfg!(target_os = "linux") {
                assert_eq!(
                    result["metrics"]["source"],
                    serde_json::json!("linux_procfs_process_tree")
                );
                assert!(result["metrics"]["sample_count"].as_u64().is_some());
            }
        });
    }

    #[test]
    fn reverse_worker_loop_backs_off_when_no_job_is_available() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            let store = JobStore::default();
            let (broker_url, handle) = spawn_mock_broker(store, 1);
            let stop = Arc::new(AtomicBool::new(false));
            let stop_after_sleep = stop.clone();
            let mut sleeps = Vec::new();
            let (output, exit_code) = run_loop(worker_options(broker_url), stop, |duration| {
                sleeps.push(duration);
                stop_after_sleep.store(true, Ordering::SeqCst);
            })
            .expect("run loop");

            assert_eq!(exit_code, 0);
            assert!(!output.claimed);
            assert!(output.stopped);
            assert_eq!(output.iterations, 1);
            assert_eq!(output.last_claim, None);
            assert_eq!(output.last_result, None);
            assert_eq!(output.last_error, None);
            assert_eq!(sleeps, vec![Duration::from_millis(1)]);
            handle.join().expect("mock broker joins");
        });
    }

    #[test]
    fn reverse_worker_reports_execution_failure_to_broker() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            let store = JobStore::default();
            store
                .submit_remote_runner_job(RemoteRunnerJobRequest {
                    runner_id: "lab".to_string(),
                    project_id: None,
                    operation: "runner.exec".to_string(),
                    command: vec!["not-allowed".to_string()],
                    cwd: Some("/tmp".to_string()),
                    env: Default::default(),
                    secret_env_names: Vec::new(),
                    capture_patch: false,
                    source_snapshot: None,
                    require_paths: Vec::new(),
                    metadata: None,
                })
                .expect("submit job");
            let (broker_url, handle) = spawn_mock_broker(store.clone(), 5);

            let (output, exit_code) =
                run_reverse_worker(worker_options(broker_url)).expect("run worker");

            assert_eq!(exit_code, 1);
            assert!(output.claimed);
            let job = output.job.expect("job");
            assert_eq!(job.status, JobStatus::Failed);
            handle.join().expect("mock broker joins");
            let events = store.events(job.id).expect("events");
            assert!(events.iter().any(|event| event.kind == JobEventKind::Error));
        });
    }

    #[test]
    fn reverse_worker_loop_reports_failed_job_status() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            let store = JobStore::default();
            store
                .submit_remote_runner_job(RemoteRunnerJobRequest {
                    runner_id: "lab".to_string(),
                    project_id: None,
                    operation: "runner.exec".to_string(),
                    command: vec!["not-allowed".to_string()],
                    cwd: Some("/tmp".to_string()),
                    env: Default::default(),
                    secret_env_names: Vec::new(),
                    capture_patch: false,
                    source_snapshot: None,
                    require_paths: Vec::new(),
                    metadata: None,
                })
                .expect("submit job");
            let (broker_url, handle) = spawn_mock_broker(store.clone(), 6);
            let stop = Arc::new(AtomicBool::new(false));
            let stop_after_sleep = stop.clone();
            let mut options = worker_options(broker_url);
            options.loop_mode = true;

            let (output, exit_code) = run_loop(options, stop, |_| {
                stop_after_sleep.store(true, Ordering::SeqCst);
            })
            .expect("run loop");

            assert_eq!(exit_code, 0);
            assert!(output.claimed);
            assert_eq!(output.jobs_claimed, 1);
            assert_eq!(output.last_result, Some(1));
            assert_eq!(output.last_error.as_deref(), Some("job exited with code 1"));
            assert!(output.last_claim.is_some());
            let job = output.job.expect("job");
            assert_eq!(job.status, JobStatus::Failed);
            handle.join().expect("mock broker joins");
        });
    }

    #[test]
    fn reverse_worker_loop_stops_without_claiming_when_stop_is_already_set() {
        let stop = Arc::new(AtomicBool::new(true));
        let (output, exit_code) = run_loop(
            worker_options("http://127.0.0.1:1".to_string()),
            stop,
            |_| panic!("worker should not sleep when already stopped"),
        )
        .expect("run loop");

        assert_eq!(exit_code, 0);
        assert!(output.stopped);
        assert_eq!(output.iterations, 0);
        assert_eq!(output.jobs_claimed, 0);
        assert_eq!(output.last_claim, None);
        assert_eq!(output.last_result, None);
        assert_eq!(output.last_error, None);
    }

    #[test]
    fn reverse_worker_skips_execution_when_claim_is_cancelled_before_start() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            let store = JobStore::default();
            store
                .submit_remote_runner_job(RemoteRunnerJobRequest {
                    runner_id: "lab".to_string(),
                    project_id: None,
                    operation: "runner.exec".to_string(),
                    command: vec![
                        "sh".to_string(),
                        "-c".to_string(),
                        "printf should-not-run".to_string(),
                    ],
                    cwd: Some("/tmp".to_string()),
                    env: Default::default(),
                    secret_env_names: Vec::new(),
                    capture_patch: false,
                    source_snapshot: None,
                    require_paths: Vec::new(),
                    metadata: None,
                })
                .expect("submit job");
            let seen_paths = Arc::new(std::sync::Mutex::new(Vec::new()));
            let (broker_url, handle) =
                spawn_cancelling_after_claim_broker(store.clone(), 2, Some(seen_paths.clone()));

            let (output, exit_code) =
                run_reverse_worker(worker_options(broker_url)).expect("run worker");

            assert_eq!(exit_code, 0);
            assert!(output.claimed);
            let job = output.job.expect("job");
            assert_eq!(job.status, JobStatus::Cancelled);
            handle.join().expect("mock broker joins");
            let seen_paths = seen_paths.lock().expect("seen paths");
            assert!(!seen_paths.iter().any(|path| path.ends_with("/events")));
            assert!(!seen_paths.iter().any(|path| path.ends_with("/finish")));
        });
    }

    #[test]
    fn reverse_worker_skips_finish_when_cancelled_after_execution() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            crate::core::runner::merge(
                Some("lab"),
                &serde_json::json!({
                    "policy": RunnerPolicy {
                        allow_raw_exec: Some(true),
                        workspace_roots: vec!["/tmp".to_string()],
                        allowed_commands: vec!["sh".to_string()],
                        ..Default::default()
                    }
                })
                .to_string(),
                &[],
            )
            .expect("set policy");
            let store = JobStore::default();
            store
                .submit_remote_runner_job(RemoteRunnerJobRequest {
                    runner_id: "lab".to_string(),
                    project_id: None,
                    operation: "runner.exec".to_string(),
                    command: vec![
                        "sh".to_string(),
                        "-c".to_string(),
                        "printf worker-ok".to_string(),
                    ],
                    cwd: Some("/tmp".to_string()),
                    env: Default::default(),
                    secret_env_names: Vec::new(),
                    capture_patch: false,
                    source_snapshot: None,
                    require_paths: Vec::new(),
                    metadata: None,
                })
                .expect("submit job");
            let seen_paths = Arc::new(std::sync::Mutex::new(Vec::new()));
            let (broker_url, handle) = spawn_cancelling_on_second_snapshot_broker(
                store.clone(),
                4,
                Some(seen_paths.clone()),
            );
            write_reverse_controller_session(&broker_url);

            let (output, exit_code) =
                run_reverse_worker(worker_options(broker_url)).expect("run worker");

            assert_eq!(exit_code, 0);
            let job = output.job.expect("job");
            assert_eq!(job.status, JobStatus::Cancelled);
            handle.join().expect("mock broker joins");
            let seen_paths = seen_paths.lock().expect("seen paths");
            assert!(!seen_paths.iter().any(|path| path.ends_with("/finish")));
            let events = store.events(job.id).expect("events");
            assert!(!events
                .iter()
                .any(|event| event.kind == JobEventKind::Result));
        });
    }

    #[test]
    fn reverse_worker_interrupts_running_job_when_broker_cancel_is_observed() {
        test_support::with_isolated_home(|_| {
            crate::core::runner::create(
                r#"{"id":"lab","kind":"local","workspace_root":"/tmp"}"#,
                false,
            )
            .expect("create runner");
            crate::core::runner::merge(
                Some("lab"),
                &serde_json::json!({
                    "policy": RunnerPolicy {
                        allow_raw_exec: Some(true),
                        workspace_roots: vec!["/tmp".to_string()],
                        allowed_commands: vec!["sh".to_string()],
                        ..Default::default()
                    }
                })
                .to_string(),
                &[],
            )
            .expect("set policy");
            let cwd = std::env::temp_dir().join(format!(
                "homeboy-reverse-worker-cancel-{}",
                std::process::id()
            ));
            std::fs::create_dir_all(&cwd).expect("create test cwd");
            let marker = cwd.join("should-not-exist");
            let store = JobStore::default();
            store
                .submit_remote_runner_job(RemoteRunnerJobRequest {
                    runner_id: "lab".to_string(),
                    project_id: None,
                    operation: "runner.exec".to_string(),
                    command: vec![
                        "sh".to_string(),
                        "-c".to_string(),
                        format!("sleep 1; touch {}", marker.display()),
                    ],
                    cwd: Some(cwd.display().to_string()),
                    env: Default::default(),
                    secret_env_names: Vec::new(),
                    capture_patch: false,
                    source_snapshot: None,
                    require_paths: Vec::new(),
                    metadata: None,
                })
                .expect("submit job");
            let seen_paths = Arc::new(std::sync::Mutex::new(Vec::new()));
            let (broker_url, handle) = spawn_cancelling_on_second_snapshot_broker(
                store.clone(),
                5,
                Some(seen_paths.clone()),
            );
            write_reverse_controller_session(&broker_url);

            let (output, exit_code) =
                run_reverse_worker(worker_options(broker_url)).expect("run worker");

            assert_eq!(exit_code, 0);
            let job = output.job.expect("job");
            assert_eq!(job.status, JobStatus::Cancelled);
            assert!(
                !marker.exists(),
                "cancelled reverse worker job left the child command running"
            );
            handle.join().expect("mock broker joins");
            let seen_paths = seen_paths.lock().expect("seen paths");
            assert!(!seen_paths.iter().any(|path| path.ends_with("/finish")));
            let events = store.events(job.id).expect("events");
            assert!(!events
                .iter()
                .any(|event| event.kind == JobEventKind::Result));
        });
    }

    #[test]
    fn reverse_worker_loop_bounds_transient_broker_failures() {
        let (broker_url, handle) = spawn_failing_broker(2);
        let mut options = worker_options(broker_url);
        options.broker_retry_limit = 1;
        let stop = Arc::new(AtomicBool::new(false));
        let mut sleeps = 0;
        let err = run_loop(options, stop, |_| {
            sleeps += 1;
        })
        .expect_err("broker failures should exceed retry budget");

        assert!(err.to_string().contains("broker request failed"));
        assert_eq!(sleeps, 1);
        handle.join().expect("mock broker joins");
    }

    fn spawn_mock_broker(
        store: JobStore,
        expected_requests: usize,
    ) -> (String, std::thread::JoinHandle<()>) {
        spawn_mock_broker_with_paths(store, expected_requests, None)
    }

    fn spawn_mock_broker_with_paths(
        store: JobStore,
        expected_requests: usize,
        seen_paths: Option<Arc<std::sync::Mutex<Vec<String>>>>,
    ) -> (String, std::thread::JoinHandle<()>) {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("listener");
        let addr = listener.local_addr().expect("addr");
        let handle = std::thread::spawn(move || {
            for _ in 0..expected_requests {
                let (mut stream, _) = listener.accept().expect("accept request");
                let request = read_request(&mut stream);
                if let Some(seen_paths) = &seen_paths {
                    seen_paths
                        .lock()
                        .expect("record request path")
                        .push(request.path.clone());
                }
                let response = handle_request(&store, &request);
                write_response(&mut stream, response);
            }
        });
        (format!("http://{addr}"), handle)
    }

    fn spawn_cancelling_after_claim_broker(
        store: JobStore,
        expected_requests: usize,
        seen_paths: Option<Arc<std::sync::Mutex<Vec<String>>>>,
    ) -> (String, std::thread::JoinHandle<()>) {
        spawn_custom_broker(store, expected_requests, seen_paths, |store, request| {
            if request.path == "/runner/jobs/claim" {
                let claim = store
                    .claim_remote_runner_job("lab", None, 30_000, None)
                    .expect("claim job");
                if let Some(claim) = &claim {
                    store
                        .cancel(claim.job.id, "user requested")
                        .expect("cancel job");
                }
                return serde_json::json!({
                    "success": true,
                    "data": { "body": { "claim": claim } }
                });
            }
            handle_request(store, request)
        })
    }

    fn spawn_cancelling_on_second_snapshot_broker(
        store: JobStore,
        expected_requests: usize,
        seen_paths: Option<Arc<std::sync::Mutex<Vec<String>>>>,
    ) -> (String, std::thread::JoinHandle<()>) {
        let snapshots = Arc::new(std::sync::Mutex::new(0_u8));
        spawn_custom_broker(
            store,
            expected_requests,
            seen_paths,
            move |store, request| {
                if let Some(job_id) = request.path.strip_prefix("/jobs/") {
                    let job_id = uuid::Uuid::parse_str(job_id).expect("job id");
                    let mut snapshots = snapshots.lock().expect("snapshot count");
                    *snapshots += 1;
                    if *snapshots == 2 {
                        store.cancel(job_id, "user requested").expect("cancel job");
                    }
                    let job = store.get(job_id).expect("job");
                    return serde_json::json!({
                        "success": true,
                        "data": { "body": { "job": job } }
                    });
                }
                handle_request(store, request)
            },
        )
    }

    fn spawn_custom_broker<F>(
        store: JobStore,
        expected_requests: usize,
        seen_paths: Option<Arc<std::sync::Mutex<Vec<String>>>>,
        mut handle: F,
    ) -> (String, std::thread::JoinHandle<()>)
    where
        F: FnMut(&JobStore, &MockRequest) -> Value + Send + 'static,
    {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("listener");
        let addr = listener.local_addr().expect("addr");
        let handle = std::thread::spawn(move || {
            for _ in 0..expected_requests {
                let (mut stream, _) = listener.accept().expect("accept request");
                let request = read_request(&mut stream);
                if let Some(seen_paths) = &seen_paths {
                    seen_paths
                        .lock()
                        .expect("record request path")
                        .push(request.path.clone());
                }
                let response = handle(&store, &request);
                write_response(&mut stream, response);
            }
        });
        (format!("http://{addr}"), handle)
    }

    fn write_reverse_controller_session(broker_url: &str) {
        let path = crate::core::paths::runner_session_file("lab").expect("session path");
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).expect("create session dir");
        }
        let session = crate::core::runner::RunnerSession {
            runner_id: "lab".to_string(),
            mode: crate::core::runner::RunnerTunnelMode::Reverse,
            role: crate::core::runner::RunnerSessionRole::Controller,
            server_id: None,
            controller_id: Some("controller".to_string()),
            broker_url: Some(broker_url.to_string()),
            remote_daemon_address: None,
            local_port: None,
            local_url: None,
            tunnel_pid: None,
            remote_daemon_pid: None,
            homeboy_version: "test".to_string(),
            homeboy_build_identity: None,
            connected_at: "2026-06-19T00:00:00Z".to_string(),
            worker_identity: None,
            worker_pid: None,
            last_seen_at: None,
        };
        std::fs::write(
            path,
            serde_json::to_string(&session).expect("serialize session"),
        )
        .expect("write session");
    }

    struct MockRequest {
        method: String,
        path: String,
        body: Value,
    }

    fn read_request(stream: &mut std::net::TcpStream) -> MockRequest {
        let mut buffer = Vec::new();
        let mut temp = [0_u8; 1024];
        let header_end = loop {
            let read = stream.read(&mut temp).expect("read request");
            assert_ne!(read, 0, "request closed before headers");
            buffer.extend_from_slice(&temp[..read]);
            if let Some(index) = find_header_end(&buffer) {
                break index;
            }
        };
        let headers = String::from_utf8_lossy(&buffer[..header_end]).to_string();
        let mut request_line = headers
            .lines()
            .next()
            .expect("request line")
            .split_whitespace();
        let method = request_line.next().expect("request method").to_string();
        let path = request_line.next().expect("request path").to_string();
        let content_length = headers
            .lines()
            .find_map(|line| line.strip_prefix("content-length: "))
            .or_else(|| {
                headers
                    .lines()
                    .find_map(|line| line.strip_prefix("Content-Length: "))
            })
            .and_then(|value| value.trim().parse::<usize>().ok())
            .unwrap_or(0);
        let body_start = header_end + 4;
        while buffer.len() < body_start + content_length {
            let read = stream.read(&mut temp).expect("read body");
            assert_ne!(read, 0, "request closed before body");
            buffer.extend_from_slice(&temp[..read]);
        }
        let body = if content_length == 0 {
            Value::Null
        } else {
            serde_json::from_slice(&buffer[body_start..body_start + content_length])
                .expect("request json")
        };
        MockRequest { method, path, body }
    }

    fn find_header_end(buffer: &[u8]) -> Option<usize> {
        buffer.windows(4).position(|window| window == b"\r\n\r\n")
    }

    fn handle_request(store: &JobStore, request: &MockRequest) -> Value {
        if request.method == "GET" {
            if let Some(job_id) = request.path.strip_prefix("/jobs/") {
                let job_id = uuid::Uuid::parse_str(job_id).expect("job id");
                let job = store.get(job_id).expect("job");
                return serde_json::json!({
                    "success": true,
                    "data": { "body": { "job": job } }
                });
            }
        }
        if request.path == "/runner/jobs/claim" {
            let claim = store
                .claim_remote_runner_job("lab", None, 30_000, None)
                .expect("claim job");
            return serde_json::json!({
                "success": true,
                "data": { "body": { "claim": claim } }
            });
        }
        if let Some(job_id) = request
            .path
            .strip_prefix("/runner/jobs/")
            .and_then(|tail| tail.strip_suffix("/events"))
        {
            let job_id = uuid::Uuid::parse_str(job_id).expect("event job id");
            let event = store
                .append_remote_runner_event(
                    job_id,
                    "lab",
                    request.body["claim_id"].as_str().expect("event claim id"),
                    JobEventKind::Progress,
                    request.body["message"].as_str().map(ToString::to_string),
                    None,
                )
                .expect("append event");
            return serde_json::json!({
                "success": true,
                "data": { "body": { "event": event } }
            });
        }
        if let Some(job_id) = request
            .path
            .strip_prefix("/runner/jobs/")
            .and_then(|tail| tail.strip_suffix("/finish"))
        {
            let job_id = uuid::Uuid::parse_str(job_id).expect("finish job id");
            let result: RemoteRunnerJobResult =
                serde_json::from_value(request.body["result"].clone()).expect("finish result");
            let job = store
                .finish_remote_runner_job(
                    job_id,
                    "lab",
                    request.body["claim_id"].as_str().expect("finish claim id"),
                    result,
                )
                .expect("finish job");
            return serde_json::json!({
                "success": true,
                "data": { "body": { "job": job } }
            });
        }
        serde_json::json!({
            "success": false,
            "error": { "message": "unknown mock path" }
        })
    }

    fn write_response(stream: &mut std::net::TcpStream, body: Value) {
        let body = body.to_string();
        let response = format!(
            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
            body.len(),
            body
        );
        stream
            .write_all(response.as_bytes())
            .expect("write response");
    }

    fn spawn_failing_broker(expected_requests: usize) -> (String, std::thread::JoinHandle<()>) {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("listener");
        let addr = listener.local_addr().expect("addr");
        let handle = std::thread::spawn(move || {
            for _ in 0..expected_requests {
                let (mut stream, _) = listener.accept().expect("accept request");
                let _ = read_request(&mut stream);
                write_response(
                    &mut stream,
                    serde_json::json!({
                        "success": false,
                        "error": { "message": "broker unavailable" }
                    }),
                );
            }
        });
        (format!("http://{addr}"), handle)
    }
}
