#[path = "../routes.rs"]
mod routes;

pub use routes::{
    first_two_deep_arrivals, grate_mother_death_script,
    story_grate_mother_death_script_after_rollback,
};
use std::io::{self, Read};
use std::path::{Path, PathBuf};

/// Shared generated-route sampling for the real 300 mm-per-`Move` contract.
///
/// Core movement admits one 3000 mm/s sample per tick and steps exactly one tick,
/// so a single `Move` advances 300 mm and several Moves are required to cross a
/// 1000 mm cell. Canonical compass here matches the wire/lexicon convention used
/// by `parse_input`: north is -Y, east is +X. Routes therefore emit one cardinal
/// input per 300 mm step, never one per path cell.
pub mod sampled_route {
    use hexwork_term::Input;
    use hexwork_worldgen::{Cell, FloorPlan, VoxelWrite, WorldPlan};
    use std::collections::{BTreeMap, BTreeSet, VecDeque};

    pub fn voxel_at(writes: &[VoxelWrite], cell: Cell) -> hexwork_core::voxel::Voxel {
        writes
            .binary_search_by_key(&(cell[2], cell[1], cell[0]), |write| {
                (write.cell[2], write.cell[1], write.cell[0])
            })
            .ok()
            .map(|index| writes[index].voxel)
            .unwrap_or_default()
    }

    /// Core solidity is `material != AIR`, so routing must read material rather
    /// than the legacy `occ` flag. Enterable water depth is the real movement
    /// policy (`MovementPolicy::default().max_water_depth == 3`; core `wet`
    /// blocks only at `fluid_amt >= 4`).
    pub fn passable(writes: &[VoxelWrite], cell: Cell) -> bool {
        let air = hexwork_core::voxel::material::AIR;
        let body = voxel_at(writes, cell);
        let head = voxel_at(writes, [cell[0], cell[1], cell[2] + 1]);
        let below = voxel_at(writes, [cell[0], cell[1], cell[2] - 1]);
        body.material == air && head.material == air && below.material != air && body.fluid_amt <= 3
    }

    /// NPCs walk to their scheduled `home`/`work` posts every tick, so a static
    /// route must treat those posts as blocked too.
    pub fn world_blockers(plan: &WorldPlan) -> BTreeSet<Cell> {
        plan.actors
            .iter()
            .flat_map(|actor| [Some(actor.position), actor.home, actor.work])
            .flatten()
            .collect()
    }

    pub fn floor_blockers(plan: &FloorPlan) -> BTreeSet<Cell> {
        plan.actors
            .iter()
            .flat_map(|actor| [Some(actor.position), actor.home, actor.work])
            .flatten()
            .collect()
    }

    /// Canonical compass word for a discrete input, matching `parse_input`.
    pub fn input_word(input: Input) -> &'static str {
        match input {
            Input::Up => "north",
            Input::Down => "south",
            Input::Left => "west",
            Input::Right => "east",
            Input::UpRight => "northeast",
            Input::UpLeft => "northwest",
            Input::DownRight => "southeast",
            Input::DownLeft => "southwest",
            Input::Confirm => "confirm",
            Input::Pickup => "pickup",
            Input::Drop => "drop",
            _ => panic!("route contains non-cardinal input: {input:?}"),
        }
    }

    fn cardinal(delta: (i32, i32)) -> Input {
        match delta {
            (0, -1) => Input::Up,
            (0, 1) => Input::Down,
            (-1, 0) => Input::Left,
            (1, 0) => Input::Right,
            other => panic!("not a cardinal route step: {other:?}"),
        }
    }

    /// Sample one cardinal leg at 300 mm per Move from an explicit millimetre
    /// position, returning the inputs and the resulting position.
    fn sample_leg(position: [i64; 2], from: Cell, to: Cell) -> (Vec<Input>, [i64; 2]) {
        let (dx, dy, input) = match (to[0] - from[0], to[1] - from[1]) {
            (0, -1) => (0i64, -1i64, Input::Up),
            (0, 1) => (0, 1, Input::Down),
            (-1, 0) => (-1, 0, Input::Left),
            (1, 0) => (1, 0, Input::Right),
            delta => panic!("not a cardinal route step: {delta:?}"),
        };
        let mut out = Vec::new();
        let mut pos = position;
        // At most four Moves cross one 1000 mm cell; bound the loop so a bug fails loudly.
        for _ in 0..5 {
            pos[0] += dx * 300;
            pos[1] += dy * 300;
            out.push(input);
            if pos[0].div_euclid(1000) as i32 == to[0] && pos[1].div_euclid(1000) as i32 == to[1] {
                break;
            }
        }
        assert!(
            pos[0].div_euclid(1000) as i32 == to[0] && pos[1].div_euclid(1000) as i32 == to[1],
            "sampled route never reached cell {to:?}"
        );
        (out, pos)
    }

    /// Sample a whole cell path from its start cell's centre.
    fn sample_path(path: &[Cell]) -> Vec<Input> {
        let mut pos = [
            i64::from(path[0][0]) * 1000 + 500,
            i64::from(path[0][1]) * 1000 + 500,
        ];
        let mut out = Vec::new();
        for pair in path.windows(2) {
            let (leg, end) = sample_leg(pos, pair[0], pair[1]);
            out.extend(leg);
            pos = end;
        }
        out
    }

    pub fn route_words(
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
    ) -> Vec<Input> {
        route_words_opt(writes, start, goal, blockers)
            .unwrap_or_else(|| panic!("no deterministic route from {start:?} to {goal:?}"))
    }

    pub fn route_words_opt(
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
    ) -> Option<Vec<Input>> {
        if start == goal {
            return Some(Vec::new());
        }
        let mut queue = VecDeque::from([start]);
        let mut previous: BTreeMap<Cell, (Cell, Input)> = BTreeMap::new();
        let mut seen = BTreeSet::from([start]);
        while let Some(cell) = queue.pop_front() {
            for delta in [(0, -1), (-1, 0), (1, 0), (0, 1)] {
                let next = [cell[0] + delta.0, cell[1] + delta.1, cell[2]];
                if seen.contains(&next)
                    || (blockers.contains(&next) && next != goal)
                    || !passable(writes, next)
                {
                    continue;
                }
                previous.insert(next, (cell, cardinal(delta)));
                if next == goal {
                    let mut path = vec![goal];
                    let mut cursor = goal;
                    while cursor != start {
                        cursor = previous[&cursor].0;
                        path.push(cursor);
                    }
                    path.reverse();
                    return Some(sample_path(&path));
                }
                seen.insert(next);
                queue.push_back(next);
            }
        }
        None
    }

    pub fn append_route(
        out: &mut String,
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
    ) {
        append_route_extra(out, writes, start, goal, blockers, 0);
    }

    /// As [`route_words`], but repeat the final cardinal step `extra` times. A
    /// blocked sample spends a tick without moving, so real walks can under-run
    /// the exact 300 mm model; extra steps push the body through the goal cell and
    /// are absorbed by whatever wall bounds it, self-healing that drift.
    pub fn route_words_extra(
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
        extra: u8,
    ) -> Vec<Input> {
        let mut words = route_words(writes, start, goal, blockers);
        if let Some(last) = words.last().copied() {
            for _ in 0..extra {
                words.push(last);
            }
        }
        words
    }

    pub fn append_route_extra(
        out: &mut String,
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
        extra: u8,
    ) {
        for input in route_words_extra(writes, start, goal, blockers, extra) {
            out.push_str("input ");
            out.push_str(input_word(input));
            out.push('\n');
        }
    }

    /// Player body is 600 mm wide by 1800 mm tall (`BodyExtents::PLAYER`), so a
    /// body offset by one 300 mm step clips the flanking wall of a one-cell
    /// dead-end mouth (the D4 hymn-card alcove at `(26,6)` is exactly this). The
    /// plain 300 mm model cannot see that clip, which is the real cause of the
    /// "stops one cell short, x pinned to the cell edge" drift.
    pub const PLAYER_HALF_MM: i64 = 300;

    fn perpendicular(input: Input) -> [i64; 3] {
        match input {
            Input::Up | Input::Down => [300, 0, 0],
            _ => [0, 300, 0],
        }
    }

    fn perpendicular_word(axis: [i64; 3]) -> Input {
        if axis[0] != 0 {
            if axis[0] > 0 {
                Input::Right
            } else {
                Input::Left
            }
        } else if axis[1] > 0 {
            Input::Down
        } else {
            Input::Up
        }
    }

    fn leg_input(from: Cell, to: Cell) -> Input {
        match (to[0] - from[0], to[1] - from[1]) {
            (0, -1) => Input::Up,
            (0, 1) => Input::Down,
            (-1, 0) => Input::Left,
            (1, 0) => Input::Right,
            delta => panic!("not a cardinal route step: {delta:?}"),
        }
    }

    /// BFS cell path (canonical compass), goal exempt from obstacles so the caller
    /// can stand on a posted cell.
    pub fn path_to(
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
    ) -> Vec<Cell> {
        if start == goal {
            return vec![start];
        }
        let mut queue = VecDeque::from([start]);
        let mut previous: BTreeMap<Cell, Cell> = BTreeMap::new();
        let mut seen = BTreeSet::from([start]);
        while let Some(cell) = queue.pop_front() {
            for delta in [(0, -1), (-1, 0), (1, 0), (0, 1)] {
                let next = [cell[0] + delta.0, cell[1] + delta.1, cell[2]];
                if seen.contains(&next)
                    || (blockers.contains(&next) && next != goal)
                    || !passable(writes, next)
                {
                    continue;
                }
                previous.insert(next, cell);
                if next == goal {
                    let mut path = vec![goal];
                    let mut cursor = goal;
                    while cursor != start {
                        cursor = previous[&cursor];
                        path.push(cursor);
                    }
                    path.reverse();
                    return path;
                }
                seen.insert(next);
                queue.push_back(next);
            }
        }
        panic!("no route from {start:?} to {goal:?}");
    }

    pub fn sample_cells(path: &[Cell]) -> Vec<Input> {
        sample_path(path)
    }

    /// Sample a cell path and return the exact modelled end millimetre position.
    pub fn sample_path_end(path: &[Cell]) -> (Vec<Input>, [i64; 2]) {
        let mut pos = [
            i64::from(path[0][0]) * 1000 + 500,
            i64::from(path[0][1]) * 1000 + 500,
        ];
        let mut out = Vec::new();
        for pair in path.windows(2) {
            let (leg, end) = sample_leg(pos, pair[0], pair[1]);
            out.extend(leg);
            pos = end;
        }
        (out, pos)
    }

    /// Enter `goal` through a one-cell mouth with the body re-centred.
    ///
    /// The player body is 600 mm wide (`BodyExtents::PLAYER`), so when both
    /// perpendicular neighbours of `goal` are solid, the body must be within
    /// ±200 mm of the goal column centre or the approach step clips a flanking
    /// wall (the D4 hymn-card alcove `(26,6)` and the `(3,10,-39)` landing are
    /// both this shape). The plain 300 mm sample can stop one cell short with the
    /// body pinned to the cell edge. Because the sample is exact, the modelled
    /// perpendicular offset is known, so a short cardinal nudge centred on the
    /// goal column is emitted before the approach. Open-cell goals are unchanged.
    pub fn route_words_centered(
        writes: &[VoxelWrite],
        start: Cell,
        goal: Cell,
        blockers: &BTreeSet<Cell>,
    ) -> Vec<Input> {
        let path = path_to(writes, start, goal, blockers);
        if path.len() < 2 {
            return Vec::new();
        }
        let pre = path[path.len() - 2];
        let approach = leg_input(pre, goal);
        let (mut words, end) = sample_path_end(&path[..path.len() - 1]);
        let (full, _) = sample_path_end(&path);
        let final_steps = full.len() - words.len();
        let p = perpendicular(approach);
        let axis = if p[0] != 0 { 0usize } else { 1usize };
        let mut neg = goal;
        neg[axis] -= 1;
        let mut pos = goal;
        pos[axis] += 1;
        let perp_solid = !passable(writes, neg) && !passable(writes, pos);
        if perp_solid {
            let center = i64::from(goal[axis]) * 1000 + 500;
            let delta = center - end[axis];
            let steps = (delta as f64 / 300.0).round() as i64;
            let word = if steps >= 0 {
                perpendicular_word(p)
            } else {
                perpendicular_word([-p[0], -p[1], -p[2]])
            };
            for _ in 0..steps.abs() {
                words.push(word);
            }
        }
        for _ in 0..final_steps.max(1) {
            words.push(approach);
        }
        // Surplus presses are only safe when the goal is a true dead-end along the
        // approach axis too; otherwise they would overshoot into the next cell.
        let (move_axis, sign) = match approach {
            Input::Up => (1usize, -1i32),
            Input::Down => (1, 1),
            Input::Left => (0, -1),
            Input::Right => (0, 1),
            _ => panic!("non-cardinal approach {approach:?}"),
        };
        let mut beyond = goal;
        beyond[move_axis] += sign;
        if perp_solid && !passable(writes, beyond) {
            for _ in 0..3 {
                words.push(approach);
            }
        }
        words
    }
}
/// Shared re-planning driver for the integration tests: every leg is re-planned
/// from the durable position so 300 mm samples, walkers leaving their posts, and
/// chasing monsters cannot strand a frozen route.
pub mod route_driver {
    use super::{sampled_route, Harness};
    use hexwork_save::{decode_run, read_file_bounded, DecodedRun, MAX_COMPRESSED_BYTES};
    use hexwork_worldgen::{Cell, VoxelWrite};
    use std::collections::BTreeSet;

    pub fn decode(h: &Harness, run_id: &str) -> DecodedRun {
        let pack = hexwork_content::RuntimePack::load(&h.content).unwrap();
        decode_run(
            &read_file_bounded(
                h.data.join("runs").join(run_id).join("run.save"),
                MAX_COMPRESSED_BYTES,
            )
            .unwrap(),
            pack.hash(),
        )
        .unwrap()
    }

    pub fn player_cell(run: &DecodedRun, player: u64) -> Cell {
        run.current
            .actors
            .iter()
            .find(|a| a.id == player)
            .unwrap()
            .position
            .cell()
    }

    /// Run one headless command batch against the managed run.
    pub fn step(h: &Harness, run_id: &str, body: &str) {
        let script = h.script("route-driver-step.commands", body);
        let mut c = h.managed(&script);
        c.arg("--load").arg(run_id);
        let out = Harness::run(c);
        assert!(
            out.status.success(),
            "stdout={} stderr={}",
            String::from_utf8_lossy(&out.stdout),
            String::from_utf8_lossy(&out.stderr)
        );
    }

    /// Bounded attempt cap: a genuinely unreachable target fails fast.
    pub const ATTEMPTS: usize = 8;

    /// One re-planning batch: route the body-aware sampler from the durable cell to
    /// `target` and run it. Decodes the run exactly once.
    pub fn walk_to(
        h: &Harness,
        run_id: &str,
        writes: &[VoxelWrite],
        blockers: &BTreeSet<Cell>,
        target: Cell,
    ) {
        for _ in 0..ATTEMPTS {
            let run = decode(h, run_id);
            let current = player_cell(&run, 1);
            if current == target {
                return;
            }
            assert_eq!(
                current[2], target[2],
                "walk_to left its floor: {current:?} -> {target:?}"
            );
            let mut body = String::new();
            for word in sampled_route::route_words_centered(writes, current, target, blockers) {
                body.push_str("input ");
                body.push_str(sampled_route::input_word(word));
                body.push('\n');
            }
            body.push_str("save\nquit\n");
            step(h, run_id, &body);
        }
        panic!("walk_to did not reach {target:?} within {ATTEMPTS} attempts");
    }
}
use std::process::{Command, ExitStatus, Output, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{sync_channel, RecvTimeoutError, SyncSender};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

static NEXT: AtomicU64 = AtomicU64::new(1);

const STDOUT_LIMIT: usize = 8 * 1024 * 1024;
const STDERR_LIMIT: usize = 1024 * 1024;
const DEFAULT_DEADLINE: Duration = Duration::from_secs(15 * 60);
const PIPE_DRAIN_DEADLINE: Duration = Duration::from_secs(5);
const OUTPUT_EVENT_CAPACITY: usize = 16;
const READ_CHUNK: usize = 8 * 1024;
const JSON_LINE_LIMIT: usize = 16 * 1024;
const STATUS_POLL_INTERVAL: Duration = Duration::from_millis(10);

#[derive(Clone, Copy)]
struct PumpLimits {
    stdout: usize,
    stderr: usize,
    deadline: Duration,
    pipe_drain_deadline: Duration,
}

impl PumpLimits {
    const DEFAULT: Self = Self {
        stdout: STDOUT_LIMIT,
        stderr: STDERR_LIMIT,
        deadline: DEFAULT_DEADLINE,
        pipe_drain_deadline: PIPE_DRAIN_DEADLINE,
    };
}

/// A durable artifact acknowledgement to observe before interrupting a child.
///
/// Observation proves that the named operation reported durable before the kill
/// request. It does not prove that the child had not already started or
/// completed its next operation by the time the parent read the acknowledgement.
struct DurableMilestone<'a> {
    code: &'a str,
    run_id: Option<&'a str>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Stream {
    Stdout,
    Stderr,
}

impl Stream {
    fn index(self) -> usize {
        match self {
            Self::Stdout => 0,
            Self::Stderr => 1,
        }
    }
}

enum OutputEvent {
    Bytes(Stream, Vec<u8>),
    Overflow(Stream),
    ReadError(Stream, io::Error),
    Eof(Stream),
}

#[derive(Debug, Eq, PartialEq)]
enum FailureKind {
    Spawn,
    Status,
    Timeout,
    StdoutOverflow,
    StderrOverflow,
    Read,
    MissingMilestone,
    PipeDrainTimeout,
    CleanupTimeout,
}

struct HarnessFailure {
    kind: FailureKind,
    detail: String,
    status: Option<ExitStatus>,
    stdout: Vec<u8>,
    stderr: Vec<u8>,
}

impl std::fmt::Display for HarnessFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "bounded child harness failed ({:?}): {}; status={:?}; stdout tail={:?}; stderr tail={:?}",
            self.kind,
            self.detail,
            self.status,
            String::from_utf8_lossy(diagnostic_tail(&self.stdout)),
            String::from_utf8_lossy(diagnostic_tail(&self.stderr)),
        )
    }
}

fn diagnostic_tail(bytes: &[u8]) -> &[u8] {
    const DIAGNOSTIC_LIMIT: usize = 4 * 1024;
    &bytes[bytes.len().saturating_sub(DIAGNOSTIC_LIMIT)..]
}

struct MilestoneScanner<'a> {
    wanted: Option<DurableMilestone<'a>>,
    line: Vec<u8>,
    discarding_oversize_line: bool,
}

impl<'a> MilestoneScanner<'a> {
    fn new(wanted: Option<DurableMilestone<'a>>) -> Self {
        Self {
            wanted,
            line: Vec::with_capacity(JSON_LINE_LIMIT.min(1024)),
            discarding_oversize_line: false,
        }
    }

    fn push(&mut self, bytes: &[u8]) -> bool {
        let Some(wanted) = self.wanted.as_ref() else {
            return false;
        };
        for &byte in bytes {
            if byte == b'\n' {
                if !self.discarding_oversize_line && milestone_matches(&self.line, wanted) {
                    self.line.clear();
                    return true;
                }
                self.line.clear();
                self.discarding_oversize_line = false;
            } else if self.discarding_oversize_line {
                continue;
            } else if self.line.len() == JSON_LINE_LIMIT {
                self.line.clear();
                self.discarding_oversize_line = true;
            } else {
                self.line.push(byte);
            }
        }
        false
    }
}

fn milestone_matches(line: &[u8], wanted: &DurableMilestone<'_>) -> bool {
    let Ok(record) = serde_json::from_slice::<serde_json::Value>(line) else {
        return false;
    };
    record
        .get("schema_version")
        .and_then(serde_json::Value::as_u64)
        == Some(1)
        && record
            .get("record_kind")
            .and_then(serde_json::Value::as_str)
            == Some("artifact_milestone")
        && record.get("code").and_then(serde_json::Value::as_str) == Some(wanted.code)
        && match wanted.run_id {
            Some(run_id) => {
                record.get("run_id").and_then(serde_json::Value::as_str) == Some(run_id)
            }
            None => true,
        }
}

pub struct Harness {
    pub root: PathBuf,
    pub home: PathBuf,
    pub cwd: PathBuf,
    pub data: PathBuf,
    pub content: PathBuf,
}

impl Harness {
    pub fn new(label: &str) -> Self {
        let serial = NEXT.fetch_add(1, Ordering::Relaxed);
        let root = PathBuf::from("/var/tmp").join(format!(
            "hexwork-cli-{label}-{}-{serial}",
            std::process::id()
        ));
        let home = root.join("home");
        let cwd = root.join("cwd");
        let data = root.join("data");
        std::fs::create_dir_all(&home).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        let content = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..");
        Self {
            root,
            home,
            cwd,
            data,
            content,
        }
    }

    pub fn script(&self, name: &str, body: &str) -> PathBuf {
        let path = self.root.join(name);
        std::fs::write(&path, format!("HEXWORK-COMMANDS 1\n{body}")).unwrap();
        path
    }

    pub fn command(&self) -> Command {
        let mut command = Command::new(env!("CARGO_BIN_EXE_hexwork"));
        command
            .current_dir(&self.cwd)
            .env("HOME", &self.home)
            .env("XDG_DATA_HOME", self.home.join("xdg"))
            .env_remove("HEXWORK_CONTENT_ROOT");
        command
    }

    pub fn managed(&self, script: &Path) -> Command {
        let mut command = self.command();
        command
            .arg("--data-dir")
            .arg(&self.data)
            .arg("--content-root")
            .arg(&self.content)
            .arg("--headless-script")
            .arg(script);
        command
    }

    pub fn run(command: Command) -> Output {
        run_bounded(command, PumpLimits::DEFAULT, None)
            .unwrap_or_else(|failure| panic!("{failure}"))
    }

    pub fn records(output: &Output) -> Vec<serde_json::Value> {
        String::from_utf8(output.stdout.clone())
            .expect("stdout UTF-8")
            .lines()
            .map(|line| serde_json::from_str(line).expect("schema JSON line"))
            .collect()
    }
}

fn run_bounded(
    mut command: Command,
    limits: PumpLimits,
    milestone: Option<DurableMilestone<'_>>,
) -> Result<Output, HarnessFailure> {
    command
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let mut child = command.spawn().map_err(|error| HarnessFailure {
        kind: FailureKind::Spawn,
        detail: error.to_string(),
        status: None,
        stdout: Vec::new(),
        stderr: Vec::new(),
    })?;
    let stdout_pipe = child.stdout.take().expect("piped child stdout");
    let stderr_pipe = child.stderr.take().expect("piped child stderr");
    let (sender, receiver) = sync_channel(OUTPUT_EVENT_CAPACITY);
    let readers = vec![
        spawn_reader(Stream::Stdout, stdout_pipe, limits.stdout, sender.clone()),
        spawn_reader(Stream::Stderr, stderr_pipe, limits.stderr, sender),
    ];

    let started = Instant::now();
    let process_deadline = started.checked_add(limits.deadline).unwrap_or(started);
    let mut stdout = Vec::with_capacity(limits.stdout.min(64 * 1024));
    let mut stderr = Vec::with_capacity(limits.stderr.min(16 * 1024));
    let mut scanner = MilestoneScanner::new(milestone);
    let waiting_for_milestone = scanner.wanted.is_some();
    let mut milestone_seen = false;
    let mut stream_done = [false; 2];
    let mut status = None;
    let mut failure: Option<(FailureKind, String)> = None;
    let mut kill_requested = false;
    let mut cleanup_deadline = None;
    let mut drain_deadline = None;

    loop {
        match child.try_wait() {
            Ok(Some(exit_status)) => {
                status = Some(exit_status);
                drain_deadline.get_or_insert_with(|| deadline_after(limits.pipe_drain_deadline));
            }
            Ok(None) => {}
            Err(error) => {
                failure.get_or_insert((FailureKind::Status, error.to_string()));
            }
        }

        let now = Instant::now();
        if status.is_none() && failure.is_none() && now >= process_deadline {
            failure = Some((
                FailureKind::Timeout,
                format!("child exceeded {:?} deadline", limits.deadline),
            ));
        }

        if let Some((_, detail)) = failure.as_mut() {
            if status.is_none() && !kill_requested {
                kill_requested = true;
                cleanup_deadline = Some(deadline_after(limits.pipe_drain_deadline));
                if let Err(error) = child.kill() {
                    detail.push_str(&format!("; kill failed: {error}"));
                }
            }
        }

        if milestone_seen && status.is_none() && !kill_requested {
            kill_requested = true;
            cleanup_deadline = Some(deadline_after(limits.pipe_drain_deadline));
            if let Err(error) = child.kill() {
                failure = Some((
                    FailureKind::Status,
                    format!("durable milestone observed but kill failed: {error}"),
                ));
            }
        }

        if status.is_some() && stream_done.iter().all(|done| *done) {
            break;
        }

        if status.is_none() && cleanup_deadline.is_some_and(|deadline| Instant::now() >= deadline) {
            let (original_kind, original_detail) = failure.unwrap_or((
                FailureKind::CleanupTimeout,
                "child did not exit after milestone-triggered kill".to_string(),
            ));
            return Err(HarnessFailure {
                kind: FailureKind::CleanupTimeout,
                detail: format!(
                    "{original_detail}; unable to reap child within {:?} (original failure: {original_kind:?})",
                    limits.pipe_drain_deadline
                ),
                status,
                stdout,
                stderr,
            });
        }

        if status.is_some() && drain_deadline.is_some_and(|deadline| Instant::now() >= deadline) {
            let (kind, detail) = failure.unwrap_or((
                FailureKind::PipeDrainTimeout,
                "stdout/stderr remained open after child exit (possibly inherited by a descendant)"
                    .to_string(),
            ));
            return Err(HarnessFailure {
                kind,
                detail: format!(
                    "{detail}; pipe drain exceeded {:?}",
                    limits.pipe_drain_deadline
                ),
                status,
                stdout,
                stderr,
            });
        }

        let wait = next_wait(process_deadline, cleanup_deadline, drain_deadline);
        match receiver.recv_timeout(wait) {
            Ok(OutputEvent::Bytes(stream, bytes)) => {
                if stream == Stream::Stdout {
                    milestone_seen |= scanner.push(&bytes);
                    stdout.extend_from_slice(&bytes);
                } else {
                    stderr.extend_from_slice(&bytes);
                }
            }
            Ok(OutputEvent::Overflow(stream)) => {
                stream_done[stream.index()] = true;
                failure.get_or_insert((
                    match stream {
                        Stream::Stdout => FailureKind::StdoutOverflow,
                        Stream::Stderr => FailureKind::StderrOverflow,
                    },
                    format!("{stream:?} exceeded its retained byte limit"),
                ));
            }
            Ok(OutputEvent::ReadError(stream, error)) => {
                stream_done[stream.index()] = true;
                failure.get_or_insert((
                    FailureKind::Read,
                    format!("failed reading {stream:?}: {error}"),
                ));
            }
            Ok(OutputEvent::Eof(stream)) => stream_done[stream.index()] = true,
            Err(RecvTimeoutError::Timeout) => {}
            Err(RecvTimeoutError::Disconnected) => thread::park_timeout(wait),
        }
    }

    for reader in readers {
        reader.join().map_err(|_| HarnessFailure {
            kind: FailureKind::Read,
            detail: "output reader thread panicked".to_string(),
            status,
            stdout: stdout.clone(),
            stderr: stderr.clone(),
        })?;
    }

    if waiting_for_milestone && !milestone_seen && failure.is_none() {
        failure = Some((
            FailureKind::MissingMilestone,
            "child exited before the requested durable milestone".to_string(),
        ));
    }
    if let Some((kind, detail)) = failure {
        return Err(HarnessFailure {
            kind,
            detail,
            status,
            stdout,
            stderr,
        });
    }

    Ok(Output {
        status: status.expect("terminal child status"),
        stdout,
        stderr,
    })
}

fn deadline_after(duration: Duration) -> Instant {
    let now = Instant::now();
    now.checked_add(duration).unwrap_or(now)
}

fn next_wait(
    process_deadline: Instant,
    cleanup_deadline: Option<Instant>,
    drain_deadline: Option<Instant>,
) -> Duration {
    let now = Instant::now();
    let poll_deadline = now.checked_add(STATUS_POLL_INTERVAL).unwrap_or(now);
    let active_cleanup_deadline = cleanup_deadline.or(drain_deadline);
    let deadline = active_cleanup_deadline
        .map(|deadline| deadline.min(poll_deadline))
        .unwrap_or_else(|| process_deadline.min(poll_deadline));
    deadline.saturating_duration_since(now)
}

fn spawn_reader<R>(
    stream: Stream,
    mut reader: R,
    limit: usize,
    sender: SyncSender<OutputEvent>,
) -> JoinHandle<()>
where
    R: Read + Send + 'static,
{
    thread::spawn(move || {
        let mut total = 0usize;
        let mut scratch = [0_u8; READ_CHUNK];
        loop {
            match reader.read(&mut scratch) {
                Ok(0) => {
                    let _ = sender.send(OutputEvent::Eof(stream));
                    return;
                }
                Ok(count) => {
                    let retained = count.min(limit.saturating_sub(total));
                    if retained != 0
                        && sender
                            .send(OutputEvent::Bytes(stream, scratch[..retained].to_vec()))
                            .is_err()
                    {
                        return;
                    }
                    total += retained;
                    if retained != count {
                        let _ = sender.send(OutputEvent::Overflow(stream));
                        return;
                    }
                }
                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                Err(error) => {
                    let _ = sender.send(OutputEvent::ReadError(stream, error));
                    return;
                }
            }
        }
    })
}

impl Drop for Harness {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.root);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;

    const TEST_DEADLINE: Duration = Duration::from_secs(5);
    const TEST_DRAIN_DEADLINE: Duration = Duration::from_secs(2);

    fn helper_command(case: &str) -> Command {
        let mut command = Command::new(std::env::current_exe().expect("current test executable"));
        command
            .arg("--exact")
            .arg("support::tests::helper_process")
            .arg("--nocapture")
            .env("HEXWORK_HARNESS_HELPER", case);
        command
    }

    fn limits(stdout: usize, stderr: usize, deadline: Duration) -> PumpLimits {
        PumpLimits {
            stdout,
            stderr,
            deadline,
            pipe_drain_deadline: TEST_DRAIN_DEADLINE,
        }
    }

    #[test]
    fn cleanup_wait_ignores_expired_process_deadline() {
        let now = Instant::now();
        let wait = next_wait(
            now.checked_sub(Duration::from_secs(1)).unwrap_or(now),
            now.checked_add(Duration::from_secs(1)),
            None,
        );
        assert!(wait > Duration::ZERO);
        assert!(wait <= STATUS_POLL_INTERVAL);
    }

    #[test]
    fn bounded_success_retains_both_streams() {
        let output = run_bounded(
            helper_command("success"),
            limits(16 * 1024, 16 * 1024, TEST_DEADLINE),
            None,
        )
        .unwrap_or_else(|failure| panic!("{failure}"));
        assert!(output.status.success());
        assert!(String::from_utf8_lossy(&output.stdout).contains("stdout-success"));
        assert!(String::from_utf8_lossy(&output.stderr).contains("stderr-success"));
    }

    #[test]
    fn timeout_kills_and_reaps_child() {
        let started = Instant::now();
        let failure = run_bounded(
            helper_command("block"),
            limits(16 * 1024, 16 * 1024, Duration::from_millis(100)),
            None,
        )
        .expect_err("blocking child must time out");
        assert_eq!(failure.kind, FailureKind::Timeout);
        assert!(failure.status.is_some(), "timed-out child was reaped");
        assert!(
            started.elapsed() < TEST_DRAIN_DEADLINE,
            "cleanup exceeded its finite drain budget"
        );
    }

    #[test]
    fn stdout_and_stderr_overflow_kill_and_reap() {
        for (case, expected) in [
            ("stdout-overflow", FailureKind::StdoutOverflow),
            ("stderr-overflow", FailureKind::StderrOverflow),
        ] {
            let failure = run_bounded(
                helper_command(case),
                limits(4096, 4096, TEST_DEADLINE),
                None,
            )
            .expect_err("oversize output must fail");
            assert_eq!(failure.kind, expected, "case {case}");
            assert!(failure.status.is_some(), "overflow child was reaped");
            assert!(failure.stdout.len() <= 4096);
            assert!(failure.stderr.len() <= 4096);
        }
    }

    #[test]
    fn stdout_and_stderr_are_drained_simultaneously() {
        let output = run_bounded(
            helper_command("simultaneous"),
            limits(512 * 1024, 512 * 1024, TEST_DEADLINE),
            None,
        )
        .unwrap_or_else(|failure| panic!("{failure}"));
        assert!(output.status.success());
        assert!(output.stdout.len() >= 256 * 1024);
        assert!(output.stderr.len() >= 256 * 1024);
    }

    #[test]
    fn durable_milestone_is_parsed_before_kill() {
        let output = run_bounded(
            helper_command("milestone"),
            limits(32 * 1024, 16 * 1024, TEST_DEADLINE),
            Some(DurableMilestone {
                code: "run-snapshot-durable",
                run_id: Some("run-7"),
            }),
        )
        .unwrap_or_else(|failure| panic!("{failure}"));
        assert!(!output.status.success(), "helper should be interrupted");
        let stdout = String::from_utf8(output.stdout).expect("helper stdout UTF-8");
        assert!(stdout.contains("\"record_kind\":\"command_ack\""));
        assert!(stdout.contains("\"run_id\":\"run-7\""));
    }

    #[test]
    fn helper_process() {
        let Ok(case) = std::env::var("HEXWORK_HARNESS_HELPER") else {
            return;
        };
        match case.as_str() {
            "success" => {
                writeln!(io::stdout(), "stdout-success").unwrap();
                writeln!(io::stderr(), "stderr-success").unwrap();
            }
            "block" => block_forever(),
            "stdout-overflow" => {
                io::stdout().write_all(&vec![b'o'; 64 * 1024]).unwrap();
                io::stdout().flush().unwrap();
                block_forever();
            }
            "stderr-overflow" => {
                io::stderr().write_all(&vec![b'e'; 64 * 1024]).unwrap();
                io::stderr().flush().unwrap();
                block_forever();
            }
            "simultaneous" => {
                let stdout = thread::spawn(|| write_repeated(io::stdout(), b'o'));
                let stderr = thread::spawn(|| write_repeated(io::stderr(), b'e'));
                stdout.join().unwrap();
                stderr.join().unwrap();
            }
            "milestone" => {
                let mut stdout = io::stdout();
                writeln!(stdout, "{{\"schema_version\":1,\"record_kind\":\"command_ack\",\"code\":\"run-snapshot-durable\",\"run_id\":\"run-7\"}}").unwrap();
                writeln!(stdout, "{{\"schema_version\":1,\"record_kind\":\"artifact_milestone\",\"code\":\"run-snapshot-durable\",\"run_id\":\"wrong-run\"}}").unwrap();
                writeln!(stdout, "{{\"schema_version\":1,\"record_kind\":\"artifact_milestone\",\"code\":\"run-snapshot-durable\",\"run_id\":\"run-7\"}}").unwrap();
                stdout.flush().unwrap();
                block_forever();
            }
            other => panic!("unknown harness helper case {other}"),
        }
    }

    fn write_repeated<W: Write>(mut writer: W, byte: u8) {
        let chunk = [byte; 8 * 1024];
        for _ in 0..32 {
            writer.write_all(&chunk).unwrap();
        }
        writer.flush().unwrap();
    }

    fn block_forever() -> ! {
        loop {
            thread::park();
        }
    }
}