//! Durable fleet inbox and run ledger.
//!
//! Stores fleet state as append-only JSONL so the manager can survive
//! restarts and reconstruct queue/worker state by replaying records.
//! Artifacts are referenced by bounded metadata; large payloads live on disk
//! and are never embedded in the ledger.

#![allow(dead_code)]

use std::collections::BTreeMap;
use std::fs::OpenOptions;
use std::io::{BufRead, Write};
use std::path::{Path, PathBuf};

use anyhow::{Context, Result};
use codewhale_protocol::fleet::*;
use serde::{Deserialize, Serialize};

const FLEET_DIR: &str = ".codewhale";
const FLEET_LEDGER_FILE: &str = "fleet.jsonl";
const PARTIAL_SUFFIX: &str = ".tmp";

/// A single append-only record in the fleet ledger.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "record", rename_all = "snake_case")]
pub enum FleetLedgerRecord {
    RunCreated {
        run: FleetRun,
    },
    RunStatusChanged {
        run_id: FleetRunId,
        status: FleetRunStatus,
        timestamp: String,
    },
    TaskEnqueued {
        entry: FleetInboxEntry,
    },
    TaskLeased {
        run_id: FleetRunId,
        task_id: String,
        worker_id: String,
        leased_at: String,
        lease_expires_at: Option<String>,
    },
    TaskCompletedOrFailed {
        run_id: FleetRunId,
        task_id: String,
        worker_id: String,
        timestamp: String,
        #[serde(default = "default_terminal_task_status")]
        status: FleetTaskLedgerStatus,
    },
    EventAppended {
        event: FleetWorkerEvent,
    },
    Heartbeat {
        worker_id: String,
        timestamp: String,
        #[serde(default)]
        cpu_percent: Option<f32>,
        #[serde(default)]
        memory_mb: Option<u64>,
    },
    ReceiptRecorded {
        receipt: FleetReceipt,
    },
    AlertSent {
        run_id: FleetRunId,
        task_id: String,
        channel: String,
        timestamp: String,
    },
}

/// Reconstructed fleet state after replaying the ledger.
#[derive(Debug, Clone, Default)]
pub struct FleetLedgerState {
    pub runs: BTreeMap<String, FleetRun>,
    pub run_status_overrides: BTreeMap<String, FleetRunStatus>,
    /// Tasks keyed by run_id:task_id.
    pub tasks: BTreeMap<String, FleetTaskState>,
    /// Worker status by worker_id.
    pub workers: BTreeMap<String, FleetWorkerStatus>,
    /// Latest heartbeat by worker_id.
    pub heartbeats: BTreeMap<String, FleetHeartbeatState>,
    /// Latest event seq per worker_id:task_id.
    pub latest_seq: BTreeMap<String, u64>,
    /// Latest event envelope per worker_id:run_id:task_id.
    pub latest_events: BTreeMap<String, FleetWorkerEvent>,
    /// Artifact events keyed by worker_id:run_id:task_id:path.
    pub artifact_events: BTreeMap<String, FleetWorkerEvent>,
    /// Restart events keyed by worker_id:run_id:task_id.
    pub restarted_events: BTreeMap<String, FleetWorkerEvent>,
    /// Escalation events keyed by worker_id:run_id:task_id.
    pub escalated_events: BTreeMap<String, FleetWorkerEvent>,
    /// Completed receipts by run_id:task_id.
    pub receipts: BTreeMap<String, FleetReceipt>,
}

#[derive(Debug, Clone)]
pub struct FleetTaskState {
    pub entry: FleetInboxEntry,
    pub status: FleetTaskLedgerStatus,
    pub leased_to: Option<String>,
    pub leased_at: Option<String>,
    pub completed_at: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FleetTaskLedgerStatus {
    Enqueued,
    Leased,
    Completed,
    Failed,
    Cancelled,
}

fn default_terminal_task_status() -> FleetTaskLedgerStatus {
    FleetTaskLedgerStatus::Completed
}

#[derive(Debug, Clone)]
pub struct FleetHeartbeatState {
    pub timestamp: String,
    pub cpu_percent: Option<f32>,
    pub memory_mb: Option<u64>,
}

/// Append-only JSONL ledger for fleet runs.
#[derive(Debug)]
pub struct FleetLedger {
    ledger_path: PathBuf,
}

impl FleetLedger {
    /// Open (or create) the ledger under `workspace/.codewhale/fleet.jsonl`.
    pub fn open(workspace: &Path) -> Result<Self> {
        let dir = workspace.join(FLEET_DIR);
        std::fs::create_dir_all(&dir)
            .with_context(|| format!("creating fleet ledger dir {}", dir.display()))?;
        let ledger_path = dir.join(FLEET_LEDGER_FILE);
        if !ledger_path.exists() {
            std::fs::write(&ledger_path, "")
                .with_context(|| format!("creating fleet ledger {}", ledger_path.display()))?;
        }
        Ok(Self { ledger_path })
    }

    pub fn path(&self) -> &Path {
        &self.ledger_path
    }

    /// Append a single record without rewriting existing ledger contents.
    fn append_record(&self, record: &FleetLedgerRecord) -> Result<()> {
        let mut line = serde_json::to_string(record).context("serializing fleet ledger record")?;
        line.push('\n');
        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.ledger_path)
            .with_context(|| format!("opening fleet ledger {}", self.ledger_path.display()))?;
        file.write_all(line.as_bytes())
            .with_context(|| format!("appending fleet ledger {}", self.ledger_path.display()))?;
        file.flush()
            .with_context(|| format!("flushing fleet ledger {}", self.ledger_path.display()))?;
        file.sync_data()
            .with_context(|| format!("syncing fleet ledger {}", self.ledger_path.display()))?;
        Ok(())
    }

    pub fn create_run(&self, run: &FleetRun) -> Result<()> {
        self.append_record(&FleetLedgerRecord::RunCreated {
            run: sanitize_run_for_ledger(run),
        })
    }

    pub fn update_run_status(
        &self,
        run_id: &FleetRunId,
        status: FleetRunStatus,
        timestamp: &str,
    ) -> Result<()> {
        self.append_record(&FleetLedgerRecord::RunStatusChanged {
            run_id: run_id.clone(),
            status,
            timestamp: timestamp.to_string(),
        })
    }

    pub fn enqueue(&self, entry: FleetInboxEntry) -> Result<()> {
        self.append_record(&FleetLedgerRecord::TaskEnqueued { entry })
    }

    /// Mark a task as leased to a worker.
    pub fn lease_task(
        &self,
        run_id: &FleetRunId,
        task_id: &str,
        worker_id: &str,
        leased_at: &str,
        lease_expires_at: Option<&str>,
    ) -> Result<()> {
        self.append_record(&FleetLedgerRecord::TaskLeased {
            run_id: run_id.clone(),
            task_id: task_id.to_string(),
            worker_id: worker_id.to_string(),
            leased_at: leased_at.to_string(),
            lease_expires_at: lease_expires_at.map(String::from),
        })
    }

    /// Mark a task as completed or failed.
    pub fn complete_or_fail_task(
        &self,
        run_id: &FleetRunId,
        task_id: &str,
        worker_id: &str,
        timestamp: &str,
    ) -> Result<()> {
        self.append_record(&FleetLedgerRecord::TaskCompletedOrFailed {
            run_id: run_id.clone(),
            task_id: task_id.to_string(),
            worker_id: worker_id.to_string(),
            timestamp: timestamp.to_string(),
            status: FleetTaskLedgerStatus::Completed,
        })
    }

    pub fn mark_task_terminal_status(
        &self,
        run_id: &FleetRunId,
        task_id: &str,
        worker_id: Option<&str>,
        timestamp: &str,
        status: FleetTaskLedgerStatus,
    ) -> Result<()> {
        self.append_record(&FleetLedgerRecord::TaskCompletedOrFailed {
            run_id: run_id.clone(),
            task_id: task_id.to_string(),
            worker_id: worker_id.unwrap_or_default().to_string(),
            timestamp: timestamp.to_string(),
            status,
        })
    }

    pub fn append_event(&self, event: FleetWorkerEvent) -> Result<()> {
        self.append_record(&FleetLedgerRecord::EventAppended { event })
    }

    pub fn heartbeat(
        &self,
        worker_id: &str,
        timestamp: &str,
        cpu_percent: Option<f32>,
        memory_mb: Option<u64>,
    ) -> Result<()> {
        self.append_record(&FleetLedgerRecord::Heartbeat {
            worker_id: worker_id.to_string(),
            timestamp: timestamp.to_string(),
            cpu_percent,
            memory_mb,
        })
    }

    pub fn record_receipt(&self, receipt: FleetReceipt) -> Result<()> {
        self.append_record(&FleetLedgerRecord::ReceiptRecorded { receipt })
    }

    pub fn record_alert(
        &self,
        run_id: &FleetRunId,
        task_id: &str,
        channel: &str,
        timestamp: &str,
    ) -> Result<()> {
        self.append_record(&FleetLedgerRecord::AlertSent {
            run_id: run_id.clone(),
            task_id: task_id.to_string(),
            channel: channel.to_string(),
            timestamp: timestamp.to_string(),
        })
    }

    /// Replay the ledger and reconstruct current state. Malformed or partial
    /// lines are skipped so an interrupted write cannot corrupt earlier state.
    pub fn rebuild_state(&self) -> Result<FleetLedgerState> {
        let mut state = FleetLedgerState::default();
        if !self.ledger_path.exists() {
            return Ok(state);
        }
        let file = std::fs::File::open(&self.ledger_path)
            .with_context(|| format!("opening ledger {}", self.ledger_path.display()))?;
        let reader = std::io::BufReader::new(file);
        for (line_no, line) in reader.lines().enumerate() {
            let line = match line {
                Ok(l) => l,
                Err(err) => {
                    tracing::warn!("fleet ledger line {} unreadable: {}", line_no + 1, err);
                    continue;
                }
            };
            if line.trim().is_empty() {
                continue;
            }
            let record: FleetLedgerRecord = match serde_json::from_str(&line) {
                Ok(r) => r,
                Err(err) => {
                    tracing::warn!(
                        "fleet ledger line {} parse error (skipping): {}",
                        line_no + 1,
                        err
                    );
                    continue;
                }
            };
            apply_record(&mut state, record);
        }
        Ok(state)
    }

    /// Claim the next available inbox task for `worker_id`. Returns the
    /// enqueued entry and appends a lease record.
    pub fn claim_next(
        &self,
        worker_id: &str,
        _worker_capabilities: &[String],
        timestamp: &str,
    ) -> Result<Option<FleetInboxEntry>> {
        let state = self.rebuild_state()?;
        // Find oldest enqueued task whose task spec (if known) matches worker
        // capabilities. For now, tasks without specs match everything.
        let candidate = state
            .tasks
            .values()
            .filter(|t| matches!(t.status, FleetTaskLedgerStatus::Enqueued))
            .map(|t| &t.entry)
            .min_by_key(|e| (e.priority, e.enqueued_at.clone()))
            .cloned();
        let Some(entry) = candidate else {
            return Ok(None);
        };
        self.lease_task(&entry.run_id, &entry.task_id, worker_id, timestamp, None)?;
        Ok(Some(entry))
    }

    /// Compact the ledger by rewriting only the records needed to reconstruct
    /// current state. This truncates history but preserves run/task/event
    /// metadata and receipts.
    pub fn compact(&self) -> Result<()> {
        let state = self.rebuild_state()?;
        let tmp_path = self.ledger_path.with_extension(PARTIAL_SUFFIX);
        let mut lines = Vec::new();
        for run in state.runs.values() {
            lines.push(serde_json::to_string(&FleetLedgerRecord::RunCreated {
                run: run.clone(),
            })?);
            if let Some(status) = state.run_status_overrides.get(&run.id.0) {
                lines.push(serde_json::to_string(
                    &FleetLedgerRecord::RunStatusChanged {
                        run_id: run.id.clone(),
                        status: status.clone(),
                        timestamp: run.updated_at.clone().unwrap_or_default(),
                    },
                )?);
            }
        }
        for task in state.tasks.values() {
            lines.push(serde_json::to_string(&FleetLedgerRecord::TaskEnqueued {
                entry: task.entry.clone(),
            })?);
            if let Some(worker) = &task.leased_to {
                lines.push(serde_json::to_string(&FleetLedgerRecord::TaskLeased {
                    run_id: task.entry.run_id.clone(),
                    task_id: task.entry.task_id.clone(),
                    worker_id: worker.clone(),
                    leased_at: task.leased_at.clone().unwrap_or_default(),
                    lease_expires_at: None,
                })?);
            }
            if matches!(
                task.status,
                FleetTaskLedgerStatus::Completed
                    | FleetTaskLedgerStatus::Failed
                    | FleetTaskLedgerStatus::Cancelled
            ) {
                lines.push(serde_json::to_string(
                    &FleetLedgerRecord::TaskCompletedOrFailed {
                        run_id: task.entry.run_id.clone(),
                        task_id: task.entry.task_id.clone(),
                        worker_id: task.leased_to.clone().unwrap_or_default(),
                        timestamp: task.completed_at.clone().unwrap_or_default(),
                        status: task.status,
                    },
                )?);
            }
        }
        let mut compacted_events = BTreeMap::new();
        for event in state.latest_events.values() {
            compacted_events.insert(compact_event_key(event), event.clone());
        }
        for event in state.artifact_events.values() {
            compacted_events.insert(compact_event_key(event), event.clone());
        }
        for event in state.restarted_events.values() {
            compacted_events.insert(compact_event_key(event), event.clone());
        }
        for event in state.escalated_events.values() {
            compacted_events.insert(compact_event_key(event), event.clone());
        }
        for event in compacted_events.values() {
            lines.push(serde_json::to_string(&FleetLedgerRecord::EventAppended {
                event: event.clone(),
            })?);
        }
        for (worker_id, heartbeat) in &state.heartbeats {
            lines.push(serde_json::to_string(&FleetLedgerRecord::Heartbeat {
                worker_id: worker_id.clone(),
                timestamp: heartbeat.timestamp.clone(),
                cpu_percent: heartbeat.cpu_percent,
                memory_mb: heartbeat.memory_mb,
            })?);
        }
        for receipt in state.receipts.values() {
            lines.push(serde_json::to_string(
                &FleetLedgerRecord::ReceiptRecorded {
                    receipt: receipt.clone(),
                },
            )?);
        }
        let mut contents = lines.join("\n");
        if !contents.is_empty() {
            contents.push('\n');
        }
        std::fs::write(&tmp_path, contents)?;
        std::fs::rename(&tmp_path, &self.ledger_path)?;
        Ok(())
    }
}

fn task_key(run_id: &str, task_id: &str) -> String {
    format!("{}:{}", run_id, task_id)
}

fn event_key(worker_id: &str, run_id: &str, task_id: &str) -> String {
    format!("{}:{}:{}", worker_id, run_id, task_id)
}

fn compact_event_key(event: &FleetWorkerEvent) -> String {
    format!(
        "{}:{}:{}:{}",
        event.worker_id, event.run_id.0, event.task_id, event.seq
    )
}

fn mark_task_terminal(
    state: &mut FleetLedgerState,
    run_id: &FleetRunId,
    task_id: &str,
    worker_id: &str,
    timestamp: &str,
    status: FleetTaskLedgerStatus,
) {
    let key = task_key(&run_id.0, task_id);
    if let Some(task) = state.tasks.get_mut(&key) {
        task.status = status;
        if !worker_id.is_empty() {
            task.leased_to = Some(worker_id.to_string());
        }
        task.completed_at = Some(timestamp.to_string());
    }
}

fn artifact_event_key(event: &FleetWorkerEvent, artifact: &FleetArtifactRef) -> String {
    format!(
        "{}:{}:{}:{}",
        event.worker_id,
        event.run_id.0,
        event.task_id,
        artifact.path.display()
    )
}

fn apply_record(state: &mut FleetLedgerState, record: FleetLedgerRecord) {
    match record {
        FleetLedgerRecord::RunCreated { run } => {
            state.runs.insert(run.id.0.clone(), run);
        }
        FleetLedgerRecord::RunStatusChanged {
            run_id,
            status,
            timestamp: _,
        } => {
            state.run_status_overrides.insert(run_id.0, status);
        }
        FleetLedgerRecord::TaskEnqueued { entry } => {
            let key = task_key(&entry.run_id.0, &entry.task_id);
            state.tasks.entry(key).or_insert_with(|| FleetTaskState {
                entry,
                status: FleetTaskLedgerStatus::Enqueued,
                leased_to: None,
                leased_at: None,
                completed_at: None,
            });
        }
        FleetLedgerRecord::TaskLeased {
            run_id,
            task_id,
            worker_id,
            leased_at,
            lease_expires_at,
        } => {
            let key = task_key(&run_id.0, &task_id);
            if let Some(task) = state.tasks.get_mut(&key) {
                task.status = FleetTaskLedgerStatus::Leased;
                task.leased_to = Some(worker_id);
                task.leased_at = Some(leased_at);
                task.entry.lease_deadline = lease_expires_at;
                task.entry.attempts = task.entry.attempts.saturating_add(1);
            }
        }
        FleetLedgerRecord::TaskCompletedOrFailed {
            run_id,
            task_id,
            worker_id,
            timestamp,
            status,
        } => {
            mark_task_terminal(state, &run_id, &task_id, &worker_id, &timestamp, status);
        }
        FleetLedgerRecord::EventAppended { event } => {
            let latest_event_key = event_key(&event.worker_id, &event.run_id.0, &event.task_id);
            if state
                .latest_seq
                .get(&latest_event_key)
                .copied()
                .is_none_or(|seq| event.seq > seq)
            {
                state.latest_seq.insert(latest_event_key.clone(), event.seq);
                state.latest_events.insert(latest_event_key, event.clone());
            }
            if let FleetWorkerEventPayload::Artifact(artifact) = &event.payload {
                state
                    .artifact_events
                    .insert(artifact_event_key(&event, artifact), event.clone());
            }
            if matches!(&event.payload, FleetWorkerEventPayload::Restarted { .. }) {
                state.restarted_events.insert(
                    event_key(&event.worker_id, &event.run_id.0, &event.task_id),
                    event.clone(),
                );
            }
            if matches!(&event.payload, FleetWorkerEventPayload::Escalated { .. }) {
                state.escalated_events.insert(
                    event_key(&event.worker_id, &event.run_id.0, &event.task_id),
                    event.clone(),
                );
            }
            // Derive worker status from lifecycle events.
            match &event.payload {
                FleetWorkerEventPayload::Leased { .. }
                | FleetWorkerEventPayload::Restarted { .. }
                | FleetWorkerEventPayload::ModelWait { .. }
                | FleetWorkerEventPayload::RunningTool { .. }
                | FleetWorkerEventPayload::Heartbeat { .. }
                | FleetWorkerEventPayload::Starting
                | FleetWorkerEventPayload::Running => {
                    state
                        .workers
                        .insert(event.worker_id.clone(), FleetWorkerStatus::Busy);
                }
                FleetWorkerEventPayload::Interrupted { .. } => {
                    state
                        .workers
                        .insert(event.worker_id.clone(), FleetWorkerStatus::Draining);
                }
                FleetWorkerEventPayload::Stale { .. } => {
                    state
                        .workers
                        .insert(event.worker_id.clone(), FleetWorkerStatus::Unhealthy);
                }
                FleetWorkerEventPayload::Completed { .. } => {
                    mark_task_terminal(
                        state,
                        &event.run_id,
                        &event.task_id,
                        &event.worker_id,
                        &event.timestamp,
                        FleetTaskLedgerStatus::Completed,
                    );
                    state
                        .workers
                        .insert(event.worker_id.clone(), FleetWorkerStatus::Online);
                }
                FleetWorkerEventPayload::Failed { .. } => {
                    mark_task_terminal(
                        state,
                        &event.run_id,
                        &event.task_id,
                        &event.worker_id,
                        &event.timestamp,
                        FleetTaskLedgerStatus::Failed,
                    );
                    state
                        .workers
                        .insert(event.worker_id.clone(), FleetWorkerStatus::Online);
                }
                FleetWorkerEventPayload::Cancelled { .. } => {
                    mark_task_terminal(
                        state,
                        &event.run_id,
                        &event.task_id,
                        &event.worker_id,
                        &event.timestamp,
                        FleetTaskLedgerStatus::Cancelled,
                    );
                    state
                        .workers
                        .insert(event.worker_id.clone(), FleetWorkerStatus::Online);
                }
                _ => {}
            }
        }
        FleetLedgerRecord::Heartbeat {
            worker_id,
            timestamp,
            cpu_percent,
            memory_mb,
        } => {
            state.heartbeats.insert(
                worker_id.clone(),
                FleetHeartbeatState {
                    timestamp,
                    cpu_percent,
                    memory_mb,
                },
            );
            if state
                .workers
                .get(&worker_id)
                .cloned()
                .unwrap_or(FleetWorkerStatus::Unknown)
                != FleetWorkerStatus::Busy
            {
                state.workers.insert(worker_id, FleetWorkerStatus::Online);
            }
        }
        FleetLedgerRecord::ReceiptRecorded { receipt } => {
            let key = task_key(&receipt.run_id.0, &receipt.task_id);
            state.receipts.insert(key, receipt);
        }
        FleetLedgerRecord::AlertSent { .. } => {
            // Alerts are audit-only for state reconstruction.
        }
    }
}

fn sanitize_run_for_ledger(run: &FleetRun) -> FleetRun {
    let mut run = run.clone();
    for task in &mut run.task_specs {
        if let Some(policy) = &mut task.alert_policy {
            for channel in &mut policy.channels {
                match channel {
                    FleetAlertChannel::Slack { webhook } => {
                        webhook.url = webhook.url.as_ref().map(|_| "<redacted>".to_string());
                    }
                    FleetAlertChannel::Webhook { endpoint } => {
                        *endpoint = FleetAlertEndpoint {
                            url: endpoint.url.as_ref().map(|_| "<redacted>".to_string()),
                            url_ref: endpoint
                                .url_ref
                                .as_ref()
                                .map(|_| FleetSecretRef::new("<redacted>")),
                            secret_ref: endpoint
                                .secret_ref
                                .as_ref()
                                .map(|_| FleetSecretRef::new("<redacted>")),
                        };
                    }
                    FleetAlertChannel::PagerDuty { routing_key, .. } => {
                        *routing_key = "<redacted>".to_string();
                    }
                }
            }
        }
    }
    run
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn sample_run(id: &str) -> FleetRun {
        FleetRun {
            id: FleetRunId::from(id),
            name: "smoke".to_string(),
            status: FleetRunStatus::Running,
            task_specs: vec![],
            worker_specs: vec![],
            labels: BTreeMap::new(),
            security_policy: None,
            created_at: "2026-06-12T17:00:00Z".to_string(),
            updated_at: None,
            completed_at: None,
        }
    }

    fn sample_entry(run_id: &str, task_id: &str) -> FleetInboxEntry {
        FleetInboxEntry {
            run_id: FleetRunId::from(run_id),
            task_id: task_id.to_string(),
            priority: 0,
            enqueued_at: "2026-06-12T17:00:00Z".to_string(),
            lease_deadline: None,
            attempts: 0,
        }
    }

    #[test]
    fn fleet_ledger_create_and_rebuild_run() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        let run = sample_run("run-1");
        ledger.create_run(&run).unwrap();
        ledger
            .update_run_status(&run.id, FleetRunStatus::Completed, "2026-06-12T18:00:00Z")
            .unwrap();

        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.runs.len(), 1);
        assert_eq!(
            state.run_status_overrides["run-1"],
            FleetRunStatus::Completed
        );
    }

    #[test]
    fn fleet_ledger_enqueue_and_claim() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        ledger.create_run(&sample_run("run-1")).unwrap();
        ledger.enqueue(sample_entry("run-1", "task-a")).unwrap();
        ledger.enqueue(sample_entry("run-1", "task-b")).unwrap();

        let claimed = ledger
            .claim_next("worker-1", &[], "2026-06-12T17:01:00Z")
            .unwrap();
        assert!(claimed.is_some());
        let claimed = claimed.unwrap();
        assert_eq!(claimed.task_id, "task-a");

        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.tasks.len(), 2);
        assert_eq!(
            state.tasks["run-1:task-a"].status,
            FleetTaskLedgerStatus::Leased
        );
        assert_eq!(
            state.tasks["run-1:task-a"].leased_to.as_deref(),
            Some("worker-1")
        );
        assert_eq!(
            state.tasks["run-1:task-b"].status,
            FleetTaskLedgerStatus::Enqueued
        );
    }

    #[test]
    fn fleet_ledger_survives_restart() {
        let tmp = TempDir::new().unwrap();
        {
            let ledger = FleetLedger::open(tmp.path()).unwrap();
            ledger.create_run(&sample_run("run-1")).unwrap();
            ledger.enqueue(sample_entry("run-1", "task-a")).unwrap();
            ledger
                .lease_task(
                    &FleetRunId::from("run-1"),
                    "task-a",
                    "worker-1",
                    "2026-06-12T17:01:00Z",
                    None,
                )
                .unwrap();
        }
        // Re-open simulates process restart.
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.runs.len(), 1);
        assert_eq!(
            state.tasks["run-1:task-a"].status,
            FleetTaskLedgerStatus::Leased
        );
    }

    #[test]
    fn fleet_ledger_skips_partial_line() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        ledger.create_run(&sample_run("run-1")).unwrap();
        // Append a truncated/invalid JSON line directly; replay should keep
        // the earlier valid record and skip only the partial line.
        let mut file = OpenOptions::new().append(true).open(ledger.path()).unwrap();
        writeln!(file, "{{\"record\":\"run_created\",\"run\":").unwrap();

        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.runs.len(), 1);
        assert!(state.runs.contains_key("run-1"));
    }

    #[test]
    fn fleet_ledger_event_and_heartbeat_reconstruct_worker_status() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        ledger.create_run(&sample_run("run-1")).unwrap();
        ledger.enqueue(sample_entry("run-1", "task-a")).unwrap();
        ledger
            .append_event(FleetWorkerEvent {
                seq: 1,
                run_id: FleetRunId::from("run-1"),
                worker_id: "worker-1".to_string(),
                task_id: "task-a".to_string(),
                timestamp: "2026-06-12T17:01:00Z".to_string(),
                payload: FleetWorkerEventPayload::Running,
                extra: BTreeMap::new(),
            })
            .unwrap();
        ledger
            .heartbeat("worker-1", "2026-06-12T17:02:00Z", Some(12.5), Some(1024))
            .unwrap();

        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.workers["worker-1"], FleetWorkerStatus::Busy);
        assert_eq!(state.heartbeats["worker-1"].cpu_percent, Some(12.5));
    }

    #[test]
    fn fleet_ledger_terminal_events_preserve_failed_and_cancelled_status() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        ledger.create_run(&sample_run("run-1")).unwrap();
        ledger
            .enqueue(sample_entry("run-1", "task-failed"))
            .unwrap();
        ledger
            .enqueue(sample_entry("run-1", "task-cancelled"))
            .unwrap();

        ledger
            .append_event(FleetWorkerEvent {
                seq: 1,
                run_id: FleetRunId::from("run-1"),
                worker_id: "worker-1".to_string(),
                task_id: "task-failed".to_string(),
                timestamp: "2026-06-12T17:03:00Z".to_string(),
                payload: FleetWorkerEventPayload::Failed {
                    reason: "test failed".to_string(),
                    recoverable: false,
                },
                extra: BTreeMap::new(),
            })
            .unwrap();
        ledger
            .append_event(FleetWorkerEvent {
                seq: 2,
                run_id: FleetRunId::from("run-1"),
                worker_id: "worker-2".to_string(),
                task_id: "task-cancelled".to_string(),
                timestamp: "2026-06-12T17:04:00Z".to_string(),
                payload: FleetWorkerEventPayload::Cancelled {
                    cancelled_by: Some("operator".to_string()),
                },
                extra: BTreeMap::new(),
            })
            .unwrap();

        let state = ledger.rebuild_state().unwrap();
        assert_eq!(
            state.tasks["run-1:task-failed"].status,
            FleetTaskLedgerStatus::Failed
        );
        assert_eq!(
            state.tasks["run-1:task-cancelled"].status,
            FleetTaskLedgerStatus::Cancelled
        );

        ledger.compact().unwrap();
        let state = ledger.rebuild_state().unwrap();
        assert_eq!(
            state.tasks["run-1:task-failed"].status,
            FleetTaskLedgerStatus::Failed
        );
        assert_eq!(
            state.tasks["run-1:task-cancelled"].status,
            FleetTaskLedgerStatus::Cancelled
        );
    }

    #[test]
    fn fleet_ledger_compact_preserves_current_state() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        ledger.create_run(&sample_run("run-1")).unwrap();
        ledger.enqueue(sample_entry("run-1", "task-a")).unwrap();
        ledger
            .lease_task(
                &FleetRunId::from("run-1"),
                "task-a",
                "worker-1",
                "2026-06-12T17:01:00Z",
                None,
            )
            .unwrap();
        ledger
            .append_event(FleetWorkerEvent {
                seq: 7,
                run_id: FleetRunId::from("run-1"),
                worker_id: "worker-1".to_string(),
                task_id: "task-a".to_string(),
                timestamp: "2026-06-12T17:01:30Z".to_string(),
                payload: FleetWorkerEventPayload::Running,
                extra: BTreeMap::new(),
            })
            .unwrap();
        ledger
            .heartbeat("worker-1", "2026-06-12T17:02:00Z", Some(12.5), Some(1024))
            .unwrap();
        ledger
            .record_receipt(FleetReceipt {
                run_id: FleetRunId::from("run-1"),
                task_id: "task-a".to_string(),
                worker_id: "worker-1".to_string(),
                completed_at: "2026-06-12T17:03:00Z".to_string(),
                result: FleetTaskResult::Pass,
                failure_kind: None,
                artifacts: vec![],
                score: None,
            })
            .unwrap();

        ledger.compact().unwrap();
        let contents = std::fs::read_to_string(ledger.path()).unwrap();
        assert!(contents.lines().count() >= 5, "{contents}");

        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.runs.len(), 1);
        assert_eq!(
            state.tasks["run-1:task-a"].status,
            FleetTaskLedgerStatus::Leased
        );
        assert_eq!(state.workers["worker-1"], FleetWorkerStatus::Busy);
        assert_eq!(state.heartbeats["worker-1"].memory_mb, Some(1024));
        assert!(state.latest_seq.values().any(|seq| *seq == 7));
        assert_eq!(state.receipts["run-1:task-a"].result, FleetTaskResult::Pass);
    }

    #[test]
    fn fleet_ledger_receipt_round_trip() {
        let tmp = TempDir::new().unwrap();
        let ledger = FleetLedger::open(tmp.path()).unwrap();
        let receipt = FleetReceipt {
            run_id: FleetRunId::from("run-1"),
            task_id: "task-a".to_string(),
            worker_id: "worker-1".to_string(),
            completed_at: "2026-06-12T17:03:00Z".to_string(),
            result: FleetTaskResult::Pass,
            failure_kind: None,
            artifacts: vec![],
            score: None,
        };
        ledger.record_receipt(receipt.clone()).unwrap();
        let state = ledger.rebuild_state().unwrap();
        assert_eq!(state.receipts["run-1:task-a"].result, FleetTaskResult::Pass);
    }
}
