kshana 0.37.0

// SPDX-License-Identifier: AGPL-3.0-only
//! The campaign runner: plan the cells, skip the ones already done, run the rest in
//! parallel, and write one content-addressed result per cell.
//!
//! A **cell** is one (recording, front-end chain, loop design). Its key is the SHA-256 of
//! the canonical JSON of everything that determines its result: the recording's SHA-256 and
//! id, the condition hash, the front-end chain's name and hash, the design's name and hash,
//! the run and scoring hashes, and the engine version. A cell's result is written to
//! `cells/<key>.json` atomically: a temporary file is renamed into place. A run that is
//! stopped part-way therefore loses only the cells that were in flight, and the next run
//! skips every cell whose file is present and carries its key.
//!
//! Pending cells that share a recording and a chain are run together. The recording is
//! read once, the chain is applied once, and every design in the group tracks every
//! expected satellite in one tracking session (`iq::track::TrackSession`, with each
//! design's lock state machine), as `iq sweep` does. When there are fewer groups than
//! workers, a group's designs are split across several work items so all cores are busy.
//! Channels in a bank never interact, so grouping changes nothing in any result.
//! `tests/iq_campaign.rs` checks this: one worker and many workers give byte-identical
//! cells.
//!
//! A cell holds no wall-clock time, host or worker. Those go to `runs.jsonl`, the one
//! output that is not deterministic.

use super::conditions::{resolve, TestConditions};
use super::detect::{self, CellMonitors, KindSpan, MonitorsSpec};
use super::hash::{canonical_hash, sha256_file, sha256_hex, CanonicalHash};
use super::score::{SatScore, SatScorer, ScoreEpoch, ScoringConfig};
use super::spec::{DataClass, FrontendSpec, LoadedCampaign};
use super::truth::{TruthCursor, TruthDoppler};
use crate::iq::acq::{acquire, samples_needed, AcqConfig, AcqResult};
use crate::iq::cli::{apply_chain, build_chain, build_code};
use crate::iq::io::batch::{default_workers, run_batch_items};
use crate::iq::io::inventory::open_recording;
use crate::iq::monitor::power::PowerMonitor;
use crate::iq::monitor::run::cn0_estimate_updates;
use crate::iq::monitor::spectral::SpectralMonitor;
use crate::iq::monitor::{EpochMonitors, MonitorEvent};
use crate::iq::signals::SignalCode;
use crate::iq::track::sink::{ChannelInfo, EpochHeader, EpochSink, EventsWriter};
use crate::iq::track::{
    ChannelInit, EpochOutput, LockEvent, LockState, SessionChannel, TrackSession,
};
use crate::iq::{Cf64, IqError, SpreadingCode};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

/// The schema tag of a cell result.
pub const CELL_SCHEMA: &str = "kshana.campaign-cell/1";

/// The schema tag of the resolved campaign (`campaign.json`).
pub const PLAN_SCHEMA: &str = "kshana.campaign-plan/1";

/// How the lock state in a cell was derived: the tracking session's lock state machine
/// (`LOCKED` counts as locked) and its `false-lock` events.
pub const LOCK_SOURCE: &str = "track-session";

/// Samples read per chunk.
const CHUNK: usize = 1 << 16;

/// The engine version stamped on every result.
pub fn engine_version() -> &'static str {
    env!("CARGO_PKG_VERSION")
}

/// How a satellite's channel was started.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Handoff {
    /// `"hint"` (stated in the test conditions), `"acquired"` or `"not-acquired"`.
    pub source: String,
    /// Starting Doppler (Hz).
    pub doppler_hz: Option<f64>,
    /// Code phase at the first sample (chips).
    pub code_phase_chips: Option<f64>,
    /// Acquisition statistic.
    pub statistic: Option<f64>,
    /// Acquisition threshold.
    pub threshold: Option<f64>,
}

/// A cell's per-update tracking output, when the campaign keeps it (`run.epochs`).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct EpochFiles {
    /// `csv`, `jsonl` or `binary` (`kshana.track-epoch/1`).
    pub format: String,
    /// The epoch file, relative to the output directory.
    pub path: String,
    /// Its SHA-256.
    pub sha256: String,
    /// Its size (bytes).
    pub bytes: u64,
    /// The lock-event file (JSON Lines), relative to the output directory.
    pub events_path: String,
    /// Its SHA-256.
    pub events_sha256: String,
}

/// One satellite of a cell.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct CellSat {
    /// How the channel was started.
    pub handoff: Handoff,
    /// The scores.
    #[serde(flatten)]
    pub score: SatScore,
}

/// A recording as stamped on a result.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct RecordingStamp {
    /// Recording id.
    pub id: String,
    /// SHA-256 of the sample data.
    pub sha256: String,
    /// Condition hash.
    pub conditions_hash: String,
}

/// A named, hashed input as stamped on a result.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct NamedHash {
    /// Name.
    pub name: String,
    /// Hash.
    pub hash: String,
}

/// The result of one cell (`kshana.campaign-cell/1`).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct CellResult {
    /// [`CELL_SCHEMA`].
    pub schema: String,
    /// The cell key.
    pub key: String,
    /// What the data is.
    pub data_class: DataClass,
    /// Engine version.
    pub engine_version: String,
    /// The recording.
    pub recording: RecordingStamp,
    /// The front-end chain.
    pub frontend: NamedHash,
    /// The loop design.
    pub design: NamedHash,
    /// Run-settings hash.
    pub run_hash: String,
    /// Scoring-settings hash.
    pub scoring_hash: String,
    /// How lock was derived ([`LOCK_SOURCE`]).
    pub lock_source: String,
    /// The C/N0 estimator behind the reported C/N0 and the degradation curve (`"m2m4"` or
    /// `"nwpr"`, the campaign's `[scoring] cn0_estimator`). Both estimates are in the cell.
    pub cn0_estimator: String,
    /// Sample rate (Hz).
    pub sample_rate_hz: f64,
    /// Samples processed.
    pub samples_processed: u64,
    /// Per-satellite results, in test-condition order.
    pub satellites: Vec<CellSat>,
    /// The per-update tracking output, when kept. Its channels are this cell's acquired
    /// satellites, in test-condition order.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub epochs: Option<EpochFiles>,
    /// The monitors' alarms and their scoring against the stated events, when the campaign
    /// runs monitors.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub monitors: Option<CellMonitors>,
}

/// One planned cell.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct PlannedCell {
    /// The cell key.
    pub key: String,
    /// Index into the plan's recordings.
    pub recording: usize,
    /// Index into the plan's front-end chains.
    pub frontend: usize,
    /// Index into the plan's designs.
    pub design: usize,
}

/// A recording in the plan.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct PlannedRecording {
    /// Recording id.
    pub id: String,
    /// The test-condition file.
    pub conditions_file: String,
    /// The recording file.
    pub path: String,
    /// SHA-256 of its sample data.
    pub sha256: String,
    /// Condition hash.
    pub conditions_hash: String,
    /// The test conditions.
    pub conditions: TestConditions,
}

/// The resolved campaign (`campaign.json`).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Plan {
    /// [`PLAN_SCHEMA`].
    pub schema: String,
    /// Campaign name.
    pub name: String,
    /// What the data is.
    pub data_class: DataClass,
    /// Engine version.
    pub engine_version: String,
    /// Recordings, by id.
    pub recordings: Vec<PlannedRecording>,
    /// Front-end chains with their hashes.
    pub frontends: Vec<(FrontendSpec, String)>,
    /// Designs, resolved (with name and hash).
    pub designs: Vec<serde_json::Value>,
    /// Run-settings hash.
    pub run_hash: String,
    /// Scoring settings (bars included).
    pub scoring: ScoringConfig,
    /// Scoring-settings hash.
    pub scoring_hash: String,
    /// The monitors, when the campaign runs them.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub monitors: Option<MonitorsSpec>,
    /// Every cell, in order (recording, chain, design).
    pub cells: Vec<PlannedCell>,
}

/// How to run a campaign.
#[derive(Clone, Debug, Default)]
pub struct RunOptions {
    /// Worker threads (0 = the campaign's setting, or every core).
    pub workers: usize,
    /// Recompute cells that are already done.
    pub no_resume: bool,
    /// Run at most this many pending cells, then stop.
    pub max_cells: Option<usize>,
    /// Plan only: report what would run.
    pub dry_run: bool,
}

/// What a run did.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct RunSummary {
    /// Campaign name.
    pub name: String,
    /// Output directory.
    pub out_dir: String,
    /// Cells in the plan.
    pub cells_total: usize,
    /// Cells found already done.
    pub cells_skipped: usize,
    /// Cells run now.
    pub cells_run: usize,
    /// Cells that failed now, with their error.
    pub cells_failed: Vec<(String, String)>,
    /// Cells still pending after this run.
    pub cells_pending: usize,
    /// Samples processed now (summed over work items).
    pub samples_processed: u64,
    /// The report digest, once every cell is done.
    pub digest: Option<String>,
}

/// A recording's SHA-256: the hash of its data file, or for several data files the
/// SHA-256 of their hex digests joined in order. File hashes are cached in
/// `<out>/hashes.json` by (path, size, modification time).
fn recording_sha(
    files: &[(PathBuf, u64)],
    cache: &Mutex<BTreeMap<String, String>>,
) -> Result<String, String> {
    let mut digests = Vec::new();
    for (p, len) in files {
        let mtime = std::fs::metadata(p)
            .and_then(|m| m.modified())
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let ck = format!("{}|{len}|{mtime}", p.display());
        let hit = cache.lock().ok().and_then(|c| c.get(&ck).cloned());
        let h = match hit {
            Some(h) => h,
            None => {
                let h = sha256_file(p)?;
                if let Ok(mut c) = cache.lock() {
                    c.insert(ck, h.clone());
                }
                h
            }
        };
        digests.push(h);
    }
    Ok(if digests.len() == 1 {
        digests.remove(0)
    } else {
        sha256_hex(digests.concat().as_bytes())
    })
}

/// Everything a cell key covers.
#[derive(Clone, Copy, Debug)]
pub struct CellKeyInputs<'a> {
    /// Recording id.
    pub recording_id: &'a str,
    /// SHA-256 of the recording's sample data.
    pub recording_sha256: &'a str,
    /// Condition hash ([`TestConditions::hash`]).
    pub conditions_hash: &'a str,
    /// Front-end chain name.
    pub frontend: &'a str,
    /// Front-end chain hash ([`FrontendSpec::hash`]).
    pub frontend_hash: &'a str,
    /// Design name.
    pub design: &'a str,
    /// Design hash ([`crate::iq::track::design::Design::hash`]).
    pub design_hash: &'a str,
    /// Run-settings hash.
    pub run_hash: &'a str,
    /// Scoring-settings hash.
    pub scoring_hash: &'a str,
    /// Data class (`synthetic` or `client-confidential`).
    pub data_class: &'a str,
    /// Engine version.
    pub engine_version: &'a str,
}

/// A cell's key: the canonical hash ([`canonical_hash`]) of every input that determines
/// its result, under the cell schema tag.
pub fn cell_key(k: &CellKeyInputs) -> CanonicalHash {
    canonical_hash(&serde_json::json!({
        "schema": CELL_SCHEMA,
        "recording_id": k.recording_id,
        "recording_sha256": k.recording_sha256,
        "conditions_hash": k.conditions_hash,
        "frontend": k.frontend,
        "frontend_hash": k.frontend_hash,
        "design": k.design,
        "design_hash": k.design_hash,
        "run_hash": k.run_hash,
        "scoring_hash": k.scoring_hash,
        "data_class": k.data_class,
        "engine_version": k.engine_version,
    }))
}

/// Resolve a loaded campaign into a plan: hash every recording (checking stated hashes)
/// and key every cell.
pub fn plan(c: &LoadedCampaign, out_dir: &Path, workers: usize) -> Result<Plan, String> {
    let cache_path = out_dir.join("hashes.json");
    let cache: BTreeMap<String, String> = std::fs::read_to_string(&cache_path)
        .ok()
        .and_then(|t| serde_json::from_str(&t).ok())
        .unwrap_or_default();
    let cache = Mutex::new(cache);
    let shas = run_batch_items(&c.conditions, workers, |(file, tc)| {
        let path = tc.recording_path(file);
        let opened = open_recording(&path, tc.recording.raw_sidecar()?)
            .map_err(|e| format!("{}: {e}", path.display()))?;
        let sha = recording_sha(&opened.data_files, &cache)?;
        if let Some(want) = &tc.recording.sha256 {
            if !want.eq_ignore_ascii_case(&sha) {
                return Err(format!(
                    "recording '{}': SHA-256 {sha} does not match the stated {want}",
                    tc.recording.id
                ));
            }
        }
        Ok((path, sha))
    });
    let mut recordings = Vec::new();
    for ((file, tc), r) in c.conditions.iter().zip(shas) {
        let (path, sha) = r?;
        recordings.push(PlannedRecording {
            id: tc.recording.id.clone(),
            conditions_file: file.display().to_string(),
            path: path.display().to_string(),
            sha256: sha,
            conditions_hash: tc.hash(),
            conditions: tc.clone(),
        });
    }
    if let Ok(cache) = cache.into_inner() {
        std::fs::create_dir_all(out_dir).map_err(|e| format!("{}: {e}", out_dir.display()))?;
        write_atomic(
            &cache_path,
            serde_json::to_string_pretty(&cache)
                .unwrap_or_default()
                .as_bytes(),
        )?;
    }
    let frontends: Vec<(FrontendSpec, String)> =
        c.frontends.iter().map(|f| (f.clone(), f.hash())).collect();
    let run_hash = c.spec.run_hash();
    let scoring_hash = c.spec.scoring_hash();
    let mut cells = Vec::new();
    for (ri, rec) in recordings.iter().enumerate() {
        for (fi, fe) in frontends.iter().enumerate() {
            for (di, d) in c.designs.iter().enumerate() {
                cells.push(PlannedCell {
                    key: cell_key(&CellKeyInputs {
                        recording_id: &rec.id,
                        recording_sha256: &rec.sha256,
                        conditions_hash: &rec.conditions_hash,
                        frontend: &fe.0.name,
                        frontend_hash: &fe.1,
                        design: d.name(),
                        design_hash: d.hash(),
                        run_hash: &run_hash,
                        scoring_hash: &scoring_hash,
                        data_class: c.spec.data_class.as_str(),
                        engine_version: engine_version(),
                    }),
                    recording: ri,
                    frontend: fi,
                    design: di,
                });
            }
        }
    }
    Ok(Plan {
        schema: PLAN_SCHEMA.into(),
        name: c.spec.name.clone(),
        data_class: c.spec.data_class,
        engine_version: engine_version().into(),
        recordings,
        frontends,
        designs: c.designs.iter().map(|d| d.to_json()).collect(),
        run_hash,
        scoring: c.spec.scoring.clone(),
        scoring_hash,
        monitors: c.spec.monitors,
        cells,
    })
}

/// Write `bytes` to `path` atomically (a temporary file in the same directory, renamed).
pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), String> {
    let dir = path.parent().unwrap_or(Path::new("."));
    std::fs::create_dir_all(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
    let tmp = dir.join(format!(
        ".{}.tmp{}",
        path.file_name()
            .map(|s| s.to_string_lossy())
            .unwrap_or_default(),
        std::process::id()
    ));
    std::fs::write(&tmp, bytes).map_err(|e| format!("{}: {e}", tmp.display()))?;
    std::fs::rename(&tmp, path).map_err(|e| format!("{}: {e}", path.display()))
}

/// The path of a cell's result.
pub fn cell_path(out_dir: &Path, key: &str) -> PathBuf {
    out_dir.join("cells").join(format!("{key}.json"))
}

/// Whether a cell's result is present and is that cell's.
pub fn cell_done(out_dir: &Path, key: &str) -> bool {
    std::fs::read_to_string(cell_path(out_dir, key))
        .ok()
        .and_then(|t| serde_json::from_str::<CellResult>(&t).ok())
        .is_some_and(|c| c.key == key)
}

/// The canonical text of a cell result: pretty JSON, keys sorted, trailing newline.
pub fn cell_text(c: &CellResult) -> String {
    let v = serde_json::to_value(c).unwrap_or_default();
    format!("{}\n", serde_json::to_string_pretty(&v).unwrap_or_default())
}

/// One work item: some designs of one (recording, chain) group.
struct WorkItem {
    recording: usize,
    frontend: usize,
    cells: Vec<usize>,
}

/// Run a loaded campaign into `out_dir`, then rebuild the report.
pub fn run(c: &LoadedCampaign, out_dir: &Path, opts: &RunOptions) -> Result<RunSummary, String> {
    let workers = match (opts.workers, c.spec.run.workers) {
        (0, 0) => default_workers(),
        (0, w) => w,
        (w, _) => w,
    };
    std::fs::create_dir_all(out_dir).map_err(|e| format!("{}: {e}", out_dir.display()))?;
    let plan = plan(c, out_dir, workers)?;
    write_atomic(
        &out_dir.join("campaign.json"),
        format!(
            "{}\n",
            serde_json::to_string_pretty(&serde_json::to_value(&plan).unwrap_or_default())
                .unwrap_or_default()
        )
        .as_bytes(),
    )?;

    let mut pending = Vec::new();
    let mut skipped = 0;
    for (i, cell) in plan.cells.iter().enumerate() {
        if !opts.no_resume && cell_done(out_dir, &cell.key) {
            skipped += 1;
        } else {
            pending.push(i);
        }
    }
    let to_run: Vec<usize> = match opts.max_cells {
        Some(m) => pending.iter().copied().take(m).collect(),
        None => pending.clone(),
    };
    let mut summary = RunSummary {
        name: plan.name.clone(),
        out_dir: out_dir.display().to_string(),
        cells_total: plan.cells.len(),
        cells_skipped: skipped,
        cells_run: 0,
        cells_failed: Vec::new(),
        cells_pending: pending.len(),
        samples_processed: 0,
        digest: None,
    };
    if opts.dry_run {
        return Ok(summary);
    }

    let items = work_items(&plan, &to_run, workers);
    let log = Mutex::new(
        std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(out_dir.join("runs.jsonl"))
            .map_err(|e| format!("runs.jsonl: {e}"))?,
    );
    let results = run_batch_items(&items, workers, |item| {
        let t0 = std::time::Instant::now();
        let r = run_item(c, &plan, item, out_dir);
        let wall = t0.elapsed().as_secs_f64();
        if let Ok(mut f) = log.lock() {
            for &ci in &item.cells {
                let cell = &plan.cells[ci];
                let line = serde_json::json!({
                    "key": cell.key,
                    "recording": plan.recordings[cell.recording].id,
                    "frontend": plan.frontends[cell.frontend].0.name,
                    "design": c.designs[cell.design].name(),
                    "item_cells": item.cells.len(),
                    "wall_s": wall,
                    "status": if r.is_ok() { "done" } else { "failed" },
                    "error": r.as_ref().err(),
                });
                let _ = writeln!(f, "{line}");
            }
        }
        r
    });
    for (item, r) in items.iter().zip(results) {
        match r {
            Ok(n) => {
                summary.cells_run += item.cells.len();
                summary.samples_processed += n;
            }
            Err(e) => {
                for &ci in &item.cells {
                    summary
                        .cells_failed
                        .push((plan.cells[ci].key.clone(), e.clone()));
                }
            }
        }
    }
    summary.cells_pending = plan
        .cells
        .iter()
        .filter(|cell| !cell_done(out_dir, &cell.key))
        .count();
    let report = super::report::build(out_dir)?;
    summary.digest = report.digest;
    Ok(summary)
}

/// Split the cells to run into work items: one per (recording, chain) group, each group
/// split into design chunks when there are fewer groups than workers.
fn work_items(plan: &Plan, to_run: &[usize], workers: usize) -> Vec<WorkItem> {
    let mut groups: BTreeMap<(usize, usize), Vec<usize>> = BTreeMap::new();
    for &ci in to_run {
        let c = &plan.cells[ci];
        groups
            .entry((c.recording, c.frontend))
            .or_default()
            .push(ci);
    }
    let per_group = workers.div_ceil(groups.len().max(1)).max(1);
    let mut items = Vec::new();
    for ((recording, frontend), cells) in groups {
        let chunks = per_group.min(cells.len());
        let size = cells.len().div_ceil(chunks);
        for chunk in cells.chunks(size) {
            items.push(WorkItem {
                recording,
                frontend,
                cells: chunk.to_vec(),
            });
        }
    }
    items
}

/// The satellites a test-condition file expects, in order.
struct Sat {
    signal: String,
    id: i64,
    code: SignalCode,
    hint: Option<(f64, f64)>,
    periods_per_bit: Option<usize>,
    truth: Option<usize>,
}

fn sats_of(tc: &TestConditions, file: &Path) -> Result<(Vec<Sat>, Vec<TruthDoppler>), String> {
    let mut sats = Vec::new();
    let mut truths = Vec::new();
    for g in &tc.expected {
        let truth = match &g.truth {
            Some(p) => {
                truths.push(TruthDoppler::load(&resolve(file, p))?);
                Some(truths.len() - 1)
            }
            None => None,
        };
        for (k, &id) in g.ids.iter().enumerate() {
            let hint = match (g.doppler_hz.get(k), g.code_phase_chips.get(k)) {
                (Some(&d), Some(&p)) => Some((d, p)),
                _ => None,
            };
            sats.push(Sat {
                signal: g.signal.clone(),
                id,
                code: build_code(&g.signal, id)?,
                hint,
                periods_per_bit: g.periods_per_bit,
                truth,
            });
        }
    }
    Ok((sats, truths))
}

/// Acquire every satellite with `cfg` over the start of the (front-end processed)
/// recording.
fn acquire_all(
    rec: &PlannedRecording,
    fe: &FrontendSpec,
    sats: &[Sat],
    cfg: &AcqConfig,
) -> Result<Vec<AcqResult>, String> {
    let tc = &rec.conditions;
    let mut opened = open_recording(Path::new(&rec.path), tc.recording.raw_sidecar()?)
        .map_err(|e| e.to_string())?;
    let spec = opened.source.spec();
    let need = sats
        .iter()
        .map(|s| samples_needed(&spec, &s.code, cfg))
        .collect::<Result<Vec<_>, _>>()?
        .into_iter()
        .max()
        .unwrap_or(0);
    let mut buf = vec![Cf64::default(); need];
    let mut got = 0;
    while got < need {
        let n = opened
            .source
            .read(&mut buf[got..])
            .map_err(|e| e.to_string())?;
        if n == 0 {
            break;
        }
        got += n;
    }
    buf.truncate(got);
    let mut chain = build_chain(&fe.params(), spec.fs_hz)?;
    apply_chain(&mut chain, &mut buf);
    sats.iter()
        .map(|s| {
            let n = samples_needed(&spec, &s.code, cfg)?;
            if buf.len() < n {
                return Err(format!(
                    "{}: recording holds {} samples, {n} needed to acquire",
                    s.code.name(),
                    buf.len()
                ));
            }
            acquire(&buf, &spec, &s.code, cfg).map(|g| g.result)
        })
        .collect()
}

/// One channel of a work item.
struct Chan {
    cell: usize,
    sat: usize,
    scorer: SatScorer,
}

/// One cell's epoch and event writers, writing to temporary files until the cell is done.
struct CellWriters {
    epochs: Box<dyn EpochSink>,
    events: EventsWriter<std::io::BufWriter<std::fs::File>>,
    tmp: (PathBuf, PathBuf),
    dest: (PathBuf, PathBuf),
}

/// Feeds the tracking session's updates, lock states and events to the scorers, and to
/// each cell's epoch writers when the campaign keeps epochs.
struct ScoreSink<'a> {
    chans: Vec<Chan>,
    /// Per session channel: its monitors (when the campaign runs them) and the times of its
    /// first and last loop update.
    mons: Vec<Option<EpochMonitors>>,
    span: Vec<Option<(f64, f64)>>,
    cursors: Vec<Option<TruthCursor<'a>>>,
    /// Session channel → (cell within the item, channel within the cell).
    route: Vec<(usize, usize)>,
    writers: Vec<Option<CellWriters>>,
}

impl EpochSink for ScoreSink<'_> {
    fn epoch(&mut self, channel: usize, e: &EpochOutput, state: LockState) -> Result<(), IqError> {
        let se = ScoreEpoch {
            t_s: e.code_epoch_s,
            t_coh_s: e.t_coh_s,
            locked: state == LockState::Locked,
            phase_lock: e.phase_lock,
            code_lock: e.code_lock,
            cn0_nwpr_dbhz: e.cn0_nwpr_dbhz,
            cn0_m2m4_dbhz: e.cn0_m2m4_dbhz,
            pll_rad: e.disc.pll_rad,
            dll_chips: e.disc.dll_chips,
            doppler_hz: e.doppler_hz,
            // A tracker false lock is scored from its event (`event` below), not by flagging
            // an update: the check moves the channel to LOST, so every later update is
            // unlocked and a flag on it was never counted.
            false_lock_flag: false,
        };
        if let Some(m) = self.mons[channel].as_mut() {
            m.push_epoch(e);
        }
        let sp = &mut self.span[channel];
        *sp = Some(sp.map_or((se.t_s, se.t_s), |(a, b)| (a.min(se.t_s), b.max(se.t_s))));
        let truth = self.cursors[channel].as_mut().map(|c| c.at(se.t_s));
        self.chans[channel].scorer.push(&se, truth);
        let (cell, local) = self.route[channel];
        if let Some(w) = self.writers[cell].as_mut() {
            w.epochs.epoch(local, e, state)?;
        }
        Ok(())
    }

    fn event(&mut self, ev: &LockEvent) -> Result<(), IqError> {
        let ch = ev.channel as usize;
        // A false lock the tracker's check found while the channel was declared LOCKED is one
        // episode at the event's time. One found during pull-in was never declared locked, and
        // the repeated firings of a channel already LOST are not new episodes.
        if ev.reason == "false-lock" && ev.from == LockState::Locked {
            if let Some(c) = self.chans.get_mut(ch) {
                c.scorer.tracker_false_lock(ev.code_epoch_s);
            }
        }
        if let Some(&(cell, local)) = self.route.get(ch) {
            if let Some(w) = self.writers[cell].as_mut() {
                let mut ev = ev.clone();
                ev.channel = local as u32;
                w.events.event(&ev)?;
            }
        }
        Ok(())
    }
}

/// Open a cell's epoch writers on temporary files in `<out>/epochs/`.
fn open_writers(
    out_dir: &Path,
    key: &str,
    kind: super::spec::EpochOutputKind,
    header: &EpochHeader,
) -> Result<Option<CellWriters>, String> {
    let Some((format, ext)) = kind.format() else {
        return Ok(None);
    };
    let dir = out_dir.join("epochs");
    std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
    let pid = std::process::id();
    let dest = (
        dir.join(format!("{key}.{ext}")),
        dir.join(format!("{key}.events.jsonl")),
    );
    let tmp = (
        dir.join(format!(".{key}.{ext}.tmp{pid}")),
        dir.join(format!(".{key}.events.jsonl.tmp{pid}")),
    );
    let create = |p: &Path| {
        std::fs::File::create(p)
            .map(std::io::BufWriter::new)
            .map_err(|e| format!("{}: {e}", p.display()))
    };
    let epochs = format
        .writer(create(&tmp.0)?, header)
        .map_err(|e| e.to_string())?;
    let events = EventsWriter::new(create(&tmp.1)?);
    Ok(Some(CellWriters {
        epochs,
        events,
        tmp,
        dest,
    }))
}

/// Flush a cell's writers, move the files into place and describe them.
fn close_writers(
    mut w: CellWriters,
    out_dir: &Path,
    kind: super::spec::EpochOutputKind,
) -> Result<EpochFiles, String> {
    w.epochs.finish().map_err(|e| e.to_string())?;
    w.events.finish().map_err(|e| e.to_string())?;
    drop(w.epochs);
    drop(w.events);
    for (t, d) in [(&w.tmp.0, &w.dest.0), (&w.tmp.1, &w.dest.1)] {
        std::fs::rename(t, d).map_err(|e| format!("{}: {e}", d.display()))?;
    }
    let rel = |p: &Path| {
        p.strip_prefix(out_dir)
            .unwrap_or(p)
            .to_string_lossy()
            .replace('\\', "/")
    };
    Ok(EpochFiles {
        format: serde_json::to_value(kind)
            .ok()
            .and_then(|v| v.as_str().map(String::from))
            .unwrap_or_default(),
        path: rel(&w.dest.0),
        sha256: sha256_file(&w.dest.0)?,
        bytes: std::fs::metadata(&w.dest.0).map(|m| m.len()).unwrap_or(0),
        events_path: rel(&w.dest.1),
        events_sha256: sha256_file(&w.dest.1)?,
    })
}

/// The label of a satellite's channel in monitor output: `<signal>:<id>`.
fn channel_label(s: &Sat) -> String {
    format!("{}:{}", s.signal, s.id)
}

/// Run one work item and write its cells; returns the samples processed.
fn run_item(
    c: &LoadedCampaign,
    plan: &Plan,
    item: &WorkItem,
    out_dir: &Path,
) -> Result<u64, String> {
    let rec = &plan.recordings[item.recording];
    let (fe, fe_hash) = &plan.frontends[item.frontend];
    let tc = &rec.conditions;
    let (sats, truths) = sats_of(tc, Path::new(&rec.conditions_file))?;
    let scoring = &plan.scoring;

    let mut opened = open_recording(Path::new(&rec.path), tc.recording.raw_sidecar()?)
        .map_err(|e| e.to_string())?;
    let spec = opened.source.spec();
    let max_samples = (c.spec.run.max_seconds > 0.0)
        .then(|| (c.spec.run.max_seconds * spec.fs_hz).round() as u64);

    // Hand-offs, per cell and satellite. One acquisition per distinct search config.
    let mut acq_cache: Vec<(AcqConfig, Vec<AcqResult>)> = Vec::new();
    let mut handoffs: Vec<Vec<(Handoff, Option<ChannelInit>)>> = Vec::new();
    for &ci in &item.cells {
        let d = &c.designs[plan.cells[ci].design];
        let mut per = Vec::new();
        for (si, s) in sats.iter().enumerate() {
            let arc: Arc<dyn SpreadingCode + Send + Sync> = Arc::new(s.code.clone());
            if let Some((dop, phase)) = s.hint {
                per.push((
                    Handoff {
                        source: "hint".into(),
                        doppler_hz: Some(dop),
                        code_phase_chips: Some(phase),
                        statistic: None,
                        threshold: None,
                    },
                    Some(ChannelInit {
                        code: arc,
                        code_phase_chips: phase,
                        doppler_hz: dop,
                        periods_per_bit: s.periods_per_bit,
                    }),
                ));
                continue;
            }
            let cfg = d.acq_config(s.code.period_s());
            let idx = match acq_cache.iter().position(|(k, _)| *k == cfg) {
                Some(i) => i,
                None => {
                    acq_cache.push((cfg, acquire_all(rec, fe, &sats, &cfg)?));
                    acq_cache.len() - 1
                }
            };
            let r = &acq_cache[idx].1[si];
            let ho = Handoff {
                source: if r.acquired {
                    "acquired"
                } else {
                    "not-acquired"
                }
                .into(),
                doppler_hz: r.acquired.then_some(r.doppler_hz),
                code_phase_chips: r.acquired.then_some(r.code_phase_chips),
                statistic: Some(r.statistic),
                threshold: Some(r.threshold),
            };
            let init = r
                .acquired
                .then(|| ChannelInit::from_acquisition(arc, r, &spec, 0, s.periods_per_bit));
            per.push((ho, init));
        }
        handoffs.push(per);
    }

    // The session: one channel per (cell, acquired satellite).
    let mut setups = Vec::new();
    let mut chans = Vec::new();
    for (k, &ci) in item.cells.iter().enumerate() {
        let d = &c.designs[plan.cells[ci].design];
        for (si, (_, init)) in handoffs[k].iter().enumerate() {
            if let Some(init) = init {
                setups.push(SessionChannel::from_design(init.clone(), d));
                chans.push(Chan {
                    cell: k,
                    sat: si,
                    scorer: SatScorer::new(
                        tc,
                        &sats[si].signal,
                        sats[si].id,
                        sats[si].code.chip_rate_hz(),
                        scoring,
                    ),
                });
            }
        }
    }
    let mut route = Vec::new();
    let mut per_cell: Vec<Vec<ChannelInfo>> = item.cells.iter().map(|_| Vec::new()).collect();
    for ch in &chans {
        let d = &c.designs[plan.cells[item.cells[ch.cell]].design];
        route.push((ch.cell, per_cell[ch.cell].len()));
        per_cell[ch.cell].push(ChannelInfo {
            code: sats[ch.sat].code.name(),
            design: d.name().into(),
            design_hash: d.hash().into(),
        });
    }
    let mut writers = Vec::new();
    for (k, infos) in per_cell.into_iter().enumerate() {
        let mut header = EpochHeader::new(infos, spec.fs_hz);
        header.recording_sha256 = Some(rec.sha256.clone());
        writers.push(open_writers(
            out_dir,
            &plan.cells[item.cells[k]].key,
            c.spec.run.epochs,
            &header,
        )?);
    }
    let mut session = TrackSession::new(spec, setups)?;
    let cursors: Vec<_> = chans
        .iter()
        .map(|ch| {
            let s = &sats[ch.sat];
            s.truth
                .and_then(|t| u32::try_from(s.id).ok().and_then(|id| truths[t].cursor(id)))
        })
        .collect();
    let mons_spec = plan.monitors;
    let mons: Vec<Option<EpochMonitors>> = chans
        .iter()
        .map(|ch| {
            mons_spec.map(|m| {
                let d = &c.designs[plan.cells[item.cells[ch.cell]].design];
                let lc = d.loop_config_for(sats[ch.sat].code.period_s());
                let mut st = m.epoch;
                if st.cn0.stride == 0 {
                    st.cn0.stride = cn0_estimate_updates(&lc);
                }
                EpochMonitors::new(&channel_label(&sats[ch.sat]), lc.spacing_chips, &st)
            })
        })
        .collect();
    let span = vec![None; chans.len()];
    let mut power = mons_spec.and_then(|m| m.power.map(|p| PowerMonitor::new(spec.fs_hz, p)));
    let mut spectral =
        mons_spec.and_then(|m| m.spectral.map(|p| SpectralMonitor::new(spec.fs_hz, p)));
    let mut sink = ScoreSink {
        mons,
        span,
        chans,
        cursors,
        route,
        writers,
    };
    let mut chain = build_chain(&fe.params(), spec.fs_hz)?;
    let mut buf = vec![Cf64::default(); CHUNK];
    let mut done = 0u64;
    loop {
        let want = match max_samples {
            Some(m) => m.saturating_sub(done).min(CHUNK as u64) as usize,
            None => CHUNK,
        };
        if want == 0 {
            break;
        }
        let n = opened
            .source
            .read(&mut buf[..want])
            .map_err(|e| e.to_string())?;
        if n == 0 {
            break;
        }
        apply_chain(&mut chain, &mut buf[..n]);
        if let Some(p) = power.as_mut() {
            p.push(&buf[..n]);
        }
        if let Some(p) = spectral.as_mut() {
            p.push(&buf[..n]);
        }
        if !sink.chans.is_empty() {
            session
                .process(&buf[..n], &mut sink)
                .map_err(|e| e.to_string())?;
        }
        done += n as u64;
    }

    // Assemble and write each cell.
    let mut scores: Vec<Vec<Option<SatScore>>> =
        item.cells.iter().map(|_| vec![None; sats.len()]).collect();
    let end_s = done as f64 / spec.fs_hz;
    // The monitors' blocks, per cell: the pre-correlation alarms are shared by the cells of
    // the item (one recording through one chain); each channel's belong to its cell.
    type Stat = Vec<(f64, f64)>;
    // Per cell: its monitors block, its alarm kinds, and each kind's statistic series.
    type Assembly = (CellMonitors, Vec<KindSpan>, Vec<Stat>);
    let mut cell_mons: Vec<Option<Assembly>> = Vec::new();
    if let Some(m) = mons_spec {
        let mut alarms: Vec<MonitorEvent> = Vec::new();
        let mut kinds: Vec<KindSpan> = Vec::new();
        let mut stats: Vec<Stat> = Vec::new();
        let series = |r: &crate::iq::monitor::MonitorReport, name: &str| -> Stat {
            r.series_named(name, None)
                .map(|s| s.t_s.iter().copied().zip(s.value.iter().copied()).collect())
                .unwrap_or_default()
        };
        let pre = |kind: &str, from_s: f64| KindSpan {
            kind: kind.into(),
            channel: None,
            sat: None,
            from_s,
            to_s: end_s,
        };
        if let (Some(p), Some(st)) = (&power, m.power) {
            alarms.extend(p.events());
            // The departure from the baseline (dB), as the test sees it.
            let r = p.report();
            let base = p.reference_db().unwrap_or(f64::NAN);
            let dep: Stat = series(&r, "power_db")
                .into_iter()
                .map(|(t, v)| (t, v - base))
                .collect();
            kinds.push(pre("power_rise", st.baseline_s));
            stats.push(dep.clone());
            kinds.push(pre("power_drop", st.baseline_s));
            stats.push(dep.into_iter().map(|(t, v)| (t, -v)).collect());
        }
        if let (Some(p), Some(st)) = (&spectral, m.spectral) {
            alarms.extend(p.events());
            let r = p.report();
            kinds.push(pre("spectral_excess", st.baseline_s));
            stats.push(series(&r, "psd_excess_db"));
            kinds.push(pre("kurtosis", st.baseline_s));
            stats.push(match p.kurtosis_reference() {
                Some((k0, sd)) => series(&r, "kurtosis")
                    .into_iter()
                    .map(|(t, v)| (t, (v - k0) / sd.max(1e-12)))
                    .collect(),
                None => Vec::new(),
            });
            // The pulse test's statistic is not kept as a series: no ROC for it.
            kinds.push(pre("pulses", st.baseline_s));
            stats.push(Vec::new());
        }
        let tests = detect::pre_correlation_tests(
            m.power
                .as_ref()
                .zip(power.as_ref())
                .map(|(st, p)| (st, p.block_len())),
            m.spectral
                .as_ref()
                .zip(spectral.as_ref())
                .map(|(st, p)| (st, p.excess_threshold_db(), p.excess_pfa_closed_form())),
        );
        for _ in &item.cells {
            cell_mons.push(Some((
                CellMonitors {
                    alarms: alarms.clone(),
                    tests: tests.clone(),
                    ..CellMonitors::default()
                },
                kinds.clone(),
                stats.clone(),
            )));
        }
        for (i, ch) in sink.chans.iter().enumerate() {
            let (Some(em), Some((first, last))) = (&sink.mons[i], sink.span[i]) else {
                continue;
            };
            let Some((cm, kinds, stats)) = cell_mons[ch.cell].as_mut() else {
                continue;
            };
            let label = channel_label(&sats[ch.sat]);
            cm.alarms.extend(em.cn0.events());
            cm.alarms.extend(em.sqm.events());
            cm.tests
                .extend(detect::channel_tests(&label, &m.epoch, em.cn0.reference()));
            let cr = em.cn0.report();
            let sr = em.sqm.report();
            let ch_series = |r: &crate::iq::monitor::MonitorReport, name: &str| -> Stat {
                r.series_named(name, Some(&label))
                    .map(|s| s.t_s.iter().copied().zip(s.value.iter().copied()).collect())
                    .unwrap_or_default()
            };
            // The SQM tests' |z|: the metric averaged over `avg_epochs`, against its baseline.
            let sqm_z = |name: &str, reference: Option<(f64, f64)>| -> Stat {
                let Some((mu, sd)) = reference else {
                    return Vec::new();
                };
                let raw = ch_series(&sr, name);
                let n = m.epoch.sqm.avg_epochs.max(1);
                let mut out = Vec::new();
                let mut sum = 0.0;
                for (i, (t, v)) in raw.iter().enumerate() {
                    sum += v;
                    if i >= n {
                        sum -= raw[i - n].1;
                    }
                    if i + 1 >= n {
                        out.push((*t, ((sum / n as f64 - mu) / sd.max(1e-12)).abs()));
                    }
                }
                out
            };
            let [dref, rref] = em.sqm.references();
            let ch_stats = [
                ch_series(&cr, "cn0_cusum_drop"),
                ch_series(&cr, "cn0_cusum_rise"),
                sqm_z("sqm_delta", dref),
                sqm_z("sqm_ratio", rref),
            ];
            for ((k, base), st) in [
                ("cn0_drop", m.epoch.cn0.baseline_s),
                ("cn0_rise", m.epoch.cn0.baseline_s),
                ("sqm_delta", m.epoch.sqm.baseline_s),
                ("sqm_ratio", m.epoch.sqm.baseline_s),
            ]
            .into_iter()
            .zip(ch_stats)
            {
                stats.push(st);
                kinds.push(KindSpan {
                    kind: k.into(),
                    channel: Some(label.clone()),
                    sat: Some(sats[ch.sat].id),
                    from_s: first + base,
                    to_s: last,
                });
            }
        }
    } else {
        cell_mons = item.cells.iter().map(|_| None).collect();
    }
    let mut cell_mons: Vec<Option<CellMonitors>> = cell_mons
        .into_iter()
        .map(|o| {
            o.map(|(mut cm, kinds, stats)| {
                cm.alarms
                    .sort_by(|a, b| a.t_alarm_s.total_cmp(&b.t_alarm_s));
                let (grace, win) =
                    mons_spec.map_or((0.0, 5.0), |m| (m.detect_grace_s, m.roc_window_s));
                let (d, f) = detect::score(&cm.alarms, &kinds, tc, grace);
                cm.detections = d;
                cm.false_alarms = f;
                cm.windows = kinds
                    .iter()
                    .zip(&stats)
                    .flat_map(|(k, st)| detect::windows(k, st, tc, grace, win))
                    .collect();
                cm
            })
        })
        .collect();
    for ch in sink.chans {
        scores[ch.cell][ch.sat] = Some(ch.scorer.finish(end_s));
    }
    let mut epoch_files: Vec<Option<EpochFiles>> = Vec::new();
    for w in sink.writers {
        epoch_files.push(match w {
            Some(w) => Some(close_writers(w, out_dir, c.spec.run.epochs)?),
            None => None,
        });
    }
    for (k, &ci) in item.cells.iter().enumerate() {
        let cell = &plan.cells[ci];
        let d = &c.designs[cell.design];
        let satellites = sats
            .iter()
            .enumerate()
            .map(|(si, s)| CellSat {
                handoff: handoffs[k][si].0.clone(),
                score: scores[k][si].take().unwrap_or_else(|| {
                    SatScorer::new(tc, &s.signal, s.id, s.code.chip_rate_hz(), scoring)
                        .finish(end_s)
                }),
            })
            .collect();
        let result = CellResult {
            schema: CELL_SCHEMA.into(),
            key: cell.key.clone(),
            data_class: plan.data_class,
            engine_version: engine_version().into(),
            recording: RecordingStamp {
                id: rec.id.clone(),
                sha256: rec.sha256.clone(),
                conditions_hash: rec.conditions_hash.clone(),
            },
            frontend: NamedHash {
                name: fe.name.clone(),
                hash: fe_hash.clone(),
            },
            design: NamedHash {
                name: d.name().into(),
                hash: d.hash().into(),
            },
            run_hash: plan.run_hash.clone(),
            scoring_hash: plan.scoring_hash.clone(),
            lock_source: LOCK_SOURCE.into(),
            cn0_estimator: scoring.cn0_estimator.name().into(),
            sample_rate_hz: spec.fs_hz,
            samples_processed: done,
            satellites,
            epochs: epoch_files[k].take(),
            monitors: cell_mons[k].take(),
        };
        write_atomic(
            &cell_path(out_dir, &cell.key),
            cell_text(&result).as_bytes(),
        )?;
    }
    Ok(done)
}