deepstrike_core/runtime/chain_validator/

mod.rs

//! P7-S5 · the chain validator, batch 1: rules C1–C4 with C7 degradation marking (P2 §5).
//!
//! Host-ops tooling, not an SDK runtime path: CI gates and incident triage run the same knife,
//! and C3 needs the deterministic transition (re-plan), which only the core can perform. The CLI
//! half is `src/bin/ds-chain-validator.rs`; this module is the verdict logic.
//!
//! Input is a journal prefix — a sequence of opaque record byte blobs. Records are grouped into
//! per-operation chain segments and every segment is judged independently. Nothing here ever
//! re-serializes a record: blobs pass through untouched, so a self-digest verdict is a verdict
//! about the bytes the host durably wrote.
//!
//! The rules, and where each one gets its teeth:
//!
//! - **C1 · chain integrity** — `record[i].previous_record_digest == digest(record[i-1])`,
//!   `step_seq` strictly +1, genesis `previous_record_digest = None`. Complete segments go
//!   through [`verify_record_chain`]; a segment with degraded hops falls back to checking every
//!   link whose digests survived.
//! - **C2 · input idempotency** — one `input_id` never yields two different records: a retry
//!   must reach the same record. Grouped per operation (the idempotency key's namespace).
//! - **C3 · causal closure** — every record's resolved effect must be reproducible by
//!   re-planning the earlier records. This is the §12.2 restore ladder's genesis leg
//!   ([`restore_operation`] with no checkpoint): chain verify + deterministic re-plan + per-step
//!   record-digest comparison. It doubles as the re-plan determinism regression gate — the
//!   direct gate for 0.2.62-class "this binary does not reproduce the history it is resuming"
//!   incidents. If C1 failed, C3 reports degraded rather than re-reporting the same break.
//! - **C4 · task lineage** — the journal-direct half: every `(task_id, attempt_id)` launch pair
//!   appears at most once (the launch token is *derived* from that pair, so a repeated pair is a
//!   reused token), and a spawn resolution names an effect the same operation published at an
//!   earlier step. The parent chain itself is not journaled; it holds structurally under C3's
//!   re-plan because an orphan spawn has no outstanding effect to resolve. The durable
//!   launch-token ledger lives in checkpoints — batch 2 territory. Both limits are named in
//!   [`ValidationReport::deferred`].
//! - **C7 · degradation** — an old-format hop (strict decode fails but the
//!   identity fields survive) degrades the checks that need the missing fields instead of
//!   failing them. A proven digest mismatch fails C1 even when identity fields survive; every degraded hop is marked on its segment's report. A blob that is not a
//!   record at all counts as unparseable input, which is an exit-code-2 condition
//!   ("evidence insufficient"), never a violation.
//!
//! ## Batch 3 · the SessionLog input plane (0.2.64 S4)
//!
//! [`validate_with_session_log`] adds a second plane: SessionLog event streams. SessionLog is
//! Evidence Truth (P6 §S) — never recovery authority, and never kernel input. Where the journal
//! plane is order-independent blobs, a session log is one file's append-ordered events, so the
//! input is a list of **streams** (one per file) whose internal order is preserved.
//!
//! The core has no typed SessionLog vocabulary (P6: the core treats SessionLog as opaque JSON),
//! so events are classified leniently: the `kind` field picks the extraction shape, missing
//! additive fields parse as absent, and both spellings of host-nested fields are accepted
//! (`route.routeId` from node, `route.route_id` from python). Unknown kinds are parseable but
//! ignored — the vocabulary evolves; only C6/C8-relevant kinds are extracted. An event that is
//! not a JSON object at all counts as unparseable (exit-code-2), exactly like the journal plane.
//!
//! C7 carries across planes: old logs simply lack `provider_attempt` / the additive fields —
//! the rules that need them degrade, never fail.

use std::collections::HashMap;

use serde::Serialize;

use crate::runtime::kernel::wire::ConfigDefaults;
use crate::runtime::kernel::wire::effect::{EffectOutcome, EffectSuccess, ProviderOutcome};
use crate::runtime::kernel::wire::record::{
    KernelRecord, NormalizedPayload, RecordError, verify_record_chain,
};
use crate::runtime::kernel::wire::restore::{RestoredOperation, restore_operation};
use crate::runtime::kernel::wire::transaction::InMemoryRecordIndex;

/// The pseudo-segment for degraded hops whose `operation_id` did not survive. Kept obviously
/// synthetic so a report reader never confuses it with a real operation.
pub const UNATTRIBUTED_SEGMENT: &str = "(unattributed)";

/// Batch-1 scope limits, surfaced verbatim on every report so a reader never mistakes a green
/// segment for a complete C4.
const DEFERRED: &[&str] = &[
    "c4.parent_chain: parent links are not journaled; an orphan spawn cannot resolve (no \
     outstanding effect), which C3's re-plan enforces structurally",
    "c4.launch_token_ledger: the durable LaunchToken ledger lives in checkpoints, so reuse \
     across different TaskLaunch payloads is a batch-2 (checkpoint input) check; the \
     journal-direct shadow — (task_id, attempt_id) pair uniqueness — is checked here",
];

/// One rule's verdict on one segment.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RuleReport {
    /// `C1`…`C4`.
    pub rule: String,
    pub verdict: Verdict,
    /// What was checked, or what broke, or why the check degraded.
    pub detail: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Verdict {
    Pass,
    Fail,
    /// C7: the check could not run to completion on this segment's evidence. Never a failure.
    Degraded,
}

/// A hop whose strict record decode failed but whose identity fields survived — the C7 marking.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct DegradedHop {
    /// Position in the validator's input, for cross-referencing the raw journal.
    pub ordinal: usize,
    pub step_seq: Option<u64>,
    /// Why the strict decode rejected the bytes.
    pub reason: String,
}

/// One operation's chain, judged independently.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct SegmentReport {
    pub operation_id: String,
    pub hops: usize,
    pub degraded_hops: Vec<DegradedHop>,
    pub rules: Vec<RuleReport>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ValidationReport {
    pub segments: Vec<SegmentReport>,
    /// Blobs that are not records at all (not JSON objects, or carrying no identity fields).
    pub unparseable_records: usize,
    /// Batch 3: SessionLog↔journal cross-verification verdicts (C6/C8). Report-scope, not
    /// per-segment: these rules join two evidence planes.
    #[serde(default)]
    pub cross_checks: Vec<RuleReport>,
    /// SessionLog event blobs that were not JSON objects at all. `0` when no session plane
    /// was handed over.
    #[serde(default)]
    pub unparseable_events: usize,
    /// `Some(count)` when SessionLog streams were handed over — the total of parseable events
    /// across every stream. `None` = journal-only validation (batch-1 mode). An explicitly
    /// provided session plane with zero parseable events is evidence-insufficient (exit 2).
    #[serde(default)]
    pub session_events: Option<usize>,
    /// Batch-scope limits a green verdict does not cover.
    pub deferred: Vec<String>,
}

impl ValidationReport {
    pub fn has_violations(&self) -> bool {
        self.segments
            .iter()
            .flat_map(|segment| segment.rules.iter())
            .chain(self.cross_checks.iter())
            .any(|rule| rule.verdict == Verdict::Fail)
    }

    /// The CLI contract (P7 §3.2): `0` all green, `1` a violation was proven, `2` the evidence
    /// was insufficient. A proven violation outranks insufficient evidence; degraded hops and
    /// deferred scope never move the code. An explicitly provided SessionLog plane that yields
    /// nothing parseable is insufficient evidence of the same kind as unparseable records.
    pub fn exit_code(&self) -> i32 {
        if self.has_violations() {
            1
        } else if self.unparseable_records > 0
            || self.unparseable_events > 0
            || self.segments.is_empty()
            || matches!(self.session_events, Some(0))
        {
            2
        } else {
            0
        }
    }
}

/// One input blob, classified. `Complete` records are self-digest-verified by construction
/// ([`KernelRecord::from_record_bytes`] cannot produce an unverified one).
enum Hop {
    Complete(KernelRecord),
    Degraded(DegradedRecord),
}

struct DegradedRecord {
    ordinal: usize,
    operation_id: Option<String>,
    input_id: Option<String>,
    step_seq: Option<u64>,
    previous_record_digest: Option<String>,
    record_digest: Option<String>,
    reason: String,
    integrity_failure: bool,
}

impl DegradedRecord {
    fn marking(&self) -> DegradedHop {
        DegradedHop {
            ordinal: self.ordinal,
            step_seq: self.step_seq,
            reason: self.reason.clone(),
        }
    }
}

impl Hop {
    fn operation_id(&self) -> Option<&str> {
        match self {
            Self::Complete(record) => Some(record.operation_id().as_str()),
            Self::Degraded(degraded) => degraded.operation_id.as_deref(),
        }
    }

    fn step_seq(&self) -> Option<u64> {
        match self {
            Self::Complete(record) => Some(record.step_seq().get()),
            Self::Degraded(degraded) => degraded.step_seq,
        }
    }
}

/// Validate a journal prefix: a sequence of opaque record byte blobs, in any order. Records
/// group into per-operation segments, each judged independently; blob order never matters
/// because the chain's own `step_seq`/digest links define the order.
pub fn validate_journal<B: AsRef<[u8]>>(blobs: &[B]) -> ValidationReport {
    validate_with_session_log(blobs, &[] as &[Vec<Vec<u8>>])
}

/// Batch 3 entry point: the journal plane plus SessionLog evidence streams. Each inner slice
/// is one session-log file's events **in append order** — unlike journal blobs, event order
/// within a stream is meaningful (a `run_started` delimits the run its following attempts
/// belong to). Streams never cross-join: fingerprint and route-stability checks are per-stream.
pub fn validate_with_session_log<J, S>(journal_blobs: &[J], session_streams: &[Vec<S>]) -> ValidationReport
where
    J: AsRef<[u8]>,
    S: AsRef<[u8]>,
{
    let (outcomes, unparseable_records) = validate_journal_plane(journal_blobs);

    let mut streams: Vec<SessionStream> = Vec::with_capacity(session_streams.len());
    for stream_blobs in session_streams {
        let mut events = Vec::with_capacity(stream_blobs.len());
        let mut unparseable_events = 0;
        for blob in stream_blobs {
            match classify_session_event(blob.as_ref()) {
                Some(event) => events.push(event),
                None => unparseable_events += 1,
            }
        }
        streams.push(SessionStream {
            events,
            unparseable_events,
        });
    }

    let session_plane_provided = !session_streams.is_empty();
    let session_events = session_plane_provided
        .then(|| streams.iter().map(|stream| stream.events.len()).sum());
    let unparseable_events = streams
        .iter()
        .map(|stream| stream.unparseable_events)
        .sum();

    // C6/C8 join the planes.
    let cross_checks = if session_plane_provided {
        let mut checks = check_c6(&streams, &outcomes);
        checks.push(check_c8(&streams, &outcomes));
        checks
    } else {
        Vec::new()
    };

    let deferred: Vec<String> = DEFERRED.iter().map(|line| (*line).to_string()).collect();

    ValidationReport {
        segments: outcomes.into_iter().map(|outcome| outcome.report).collect(),
        unparseable_records,
        cross_checks,
        unparseable_events,
        session_events,
        deferred,
    }
}

/// The journal plane on its own: classify blobs into hops, group into segments, judge each.
fn validate_journal_plane<B: AsRef<[u8]>>(blobs: &[B]) -> (Vec<SegmentOutcome>, usize) {
    let mut hops: Vec<Hop> = Vec::with_capacity(blobs.len());
    let mut unparseable_records = 0;
    for (ordinal, blob) in blobs.iter().enumerate() {
        match classify(ordinal, blob.as_ref()) {
            Some(hop) => hops.push(hop),
            None => unparseable_records += 1,
        }
    }

    let mut segments: HashMap<String, Vec<Hop>> = HashMap::new();
    for hop in hops {
        let key = hop
            .operation_id()
            .map(str::to_string)
            .unwrap_or_else(|| UNATTRIBUTED_SEGMENT.to_string());
        segments.entry(key).or_default().push(hop);
    }

    let mut keys: Vec<String> = segments.keys().cloned().collect();
    keys.sort();
    let reports = keys
        .iter()
        .map(|key| validate_segment(key, segments.remove(key).unwrap_or_default()))
        .collect();
    (reports, unparseable_records)
}

// ---------------------------------------------------------------------------------------------
// batch 3 · the SessionLog evidence plane
// ---------------------------------------------------------------------------------------------

/// One session-log file, classified: its events in append order plus the count of blobs that
/// were not JSON objects at all.
pub struct SessionStream {
    pub events: Vec<EvidenceEvent>,
    pub unparseable_events: usize,
}

/// A SessionLog event, leniently classified. Only the kinds C6/C8 read are extracted; every
/// other kind — known or future — is `Other`. Field absence is data (C7 degrades), not error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EvidenceEvent {
    /// `run_started` — delimits a run; its `route` is the batch-3 route-stability baseline
    /// (Q3). Absent route = old log = degraded, never failed.
    RunStarted { route_id: Option<String> },
    /// `provider_attempt` — one effect's physical execution (P4 §1.2).
    ProviderAttempt {
        effect_id: Option<String>,
        request_fingerprint: Option<String>,
        route_id: Option<String>,
        status: Option<String>,
    },
    /// `prompt_measured` — the durable measurement fact; `request_fingerprint` joins a
    /// `provider_attempt` to the request plan it executed (G2, C6).
    PromptMeasured {
        effect_id: Option<String>,
        request_fingerprint: Option<String>,
    },
    /// `llm_completed` — the invocation's terminal projection: `invocation_id` derives as the
    /// chain's first effect (P4 §1.1), `effect_id` is the selected outcome effect.
    LlmCompleted {
        effect_id: Option<String>,
        invocation_id: Option<String>,
    },
    /// Any other kind — parseable, ignored by batch-3 rules.
    Other,
}

/// Lenient event classification: a JSON object with an extractable `kind` classifies; anything
/// else is unparseable input (exit-code-2, never a violation).
fn classify_session_event(bytes: &[u8]) -> Option<EvidenceEvent> {
    let value: serde_json::Value = serde_json::from_slice(bytes).ok()?;
    let object = value.as_object()?;
    let string = |key: &str| {
        object
            .get(key)
            .and_then(serde_json::Value::as_str)
            .map(str::to_string)
    };
    let kind = string("kind");
    let event = match kind.as_deref() {
        Some("run_started") => EvidenceEvent::RunStarted {
            route_id: route_id_of(&value),
        },
        Some("provider_attempt") => EvidenceEvent::ProviderAttempt {
            effect_id: string("effect_id"),
            request_fingerprint: string("request_fingerprint"),
            route_id: route_id_of(&value),
            status: string("status"),
        },
        Some("prompt_measured") => EvidenceEvent::PromptMeasured {
            effect_id: string("effect_id"),
            // The nested measurement keeps its host-native shape: node serializes camelCase
            // (`requestFingerprint`), python snake_case (`request_fingerprint`).
            request_fingerprint: object
                .get("measurement")
                .and_then(|measurement| {
                    measurement
                        .get("requestFingerprint")
                        .or_else(|| measurement.get("request_fingerprint"))
                })
                .and_then(serde_json::Value::as_str)
                .map(str::to_string),
        },
        Some("llm_completed") => EvidenceEvent::LlmCompleted {
            effect_id: string("effect_id"),
            invocation_id: string("invocation_id"),
        },
        _ => EvidenceEvent::Other,
    };
    Some(event)
}

/// `route.route_id`, accepting both host spellings (node camelCase, python snake_case).
fn route_id_of(event: &serde_json::Value) -> Option<String> {
    let route = event.get("route")?;
    route
        .get("routeId")
        .or_else(|| route.get("route_id"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_string)
}

// ---------------------------------------------------------------------------------------------
// C6 · SessionLog↔journal cross-verification (batch 3, 0.2.64 S4b)
// ---------------------------------------------------------------------------------------------

/// C6 joins the two planes: host-side SessionLog evidence against the journal's authority.
/// Three clauses, three reports — their degradation conditions differ, so one merged verdict
/// would hide which clause actually ran:
///
/// - **C6.1 · attempt↔journal effect correspondence** — every `provider_attempt.effect_id`
///   must name an effect its operation's segment actually published. Membership comes from
///   the deterministic re-plan (resolved effect ids read journal-directly ∪ pending effects
///   after restore): a complete segment either published the effect or it did not, which makes
///   a mismatched attempt provably forged rather than merely unverifiable. Journal prefixes
///   degrade honestly: an attempt naming a step past the journal's tip, a missing segment
///   (the journal may cover a subset of the session's operations), or an unrestorable segment
///   all degrade instead of failing.
/// - **C6.2 · fingerprint join** — every `provider_attempt.request_fingerprint` must appear on
///   a `prompt_measured` in the same stream (G2: the fingerprint binds the evidence to the
///   request plan; P4 §5). Streams never cross-join.
/// - **C6.3 · route stability** (裁决 Q3) — within one run (delimited by `run_started`), every
///   attempt's routeId equals the pinning `run_started.route.route_id`; an in-run mismatch is
///   a violation. A *new* `run_started` naming a different route is a legal cross-resume change
///   (adapter upgrades happen) and is degraded-marked, never failed.
///
/// C7 spans all three: a stream with no `provider_attempt` events at all (a pre-0.2.63 log)
/// degrades every clause. A `provider_attempt` missing its primary key (`effect_id`) or its
/// `request_fingerprint` fails — the event kind itself is new, so no old log can produce one,
/// and the conformant writers require both fields; a keyless attempt is forged evidence.
fn check_c6(streams: &[SessionStream], outcomes: &[SegmentOutcome]) -> Vec<RuleReport> {
    let segments: HashMap<&str, &SegmentOutcome> = outcomes
        .iter()
        .filter(|outcome| outcome.report.operation_id != UNATTRIBUTED_SEGMENT)
        .map(|outcome| (outcome.report.operation_id.as_str(), outcome))
        .collect();

    let mut correspondence = ClauseAccumulator::default();
    let mut fingerprints = ClauseAccumulator::default();
    let mut routes = ClauseAccumulator::default();
    let mut total_attempts = 0usize;

    for (index, stream) in streams.iter().enumerate() {
        let measured: std::collections::HashSet<&str> = stream
            .events
            .iter()
            .filter_map(|event| match event {
                EvidenceEvent::PromptMeasured {
                    request_fingerprint,
                    ..
                } => request_fingerprint.as_deref(),
                _ => None,
            })
            .collect();

        // C6.3 per-stream walk state: the current run's pinned route, and the previous run's
        // for the cross-resume comparison.
        let mut baseline: Option<&str> = None;
        let mut last_pinned: Option<&str> = None;

        for event in &stream.events {
            match event {
                EvidenceEvent::RunStarted { route_id } => {
                    if let Some(new_route) = route_id.as_deref() {
                        if let Some(previous) = last_pinned
                            && previous != new_route
                        {
                            routes.degraded(format!(
                                "stream #{index}: run resumed on route {new_route} (was \
                                 {previous}) — a cross-resume change, degraded per Q3"
                            ));
                        }
                        baseline = Some(new_route);
                        last_pinned = Some(new_route);
                    } else {
                        // A routeless run_started is old-format: attempts under it cannot be
                        // route-checked.
                        baseline = None;
                    }
                }
                EvidenceEvent::ProviderAttempt {
                    effect_id,
                    request_fingerprint,
                    route_id,
                    ..
                } => {
                    total_attempts += 1;
                    let label = effect_id.as_deref().unwrap_or("(no effect_id)");

                    // C6.1
                    match effect_id.as_deref() {
                        None => correspondence.violation(format!(
                            "stream #{index}: provider_attempt without effect_id — the writers \
                             mint it from the kernel effect, so a keyless attempt is forged \
                             evidence"
                        )),
                        Some(effect) => match parse_effect_step(effect) {
                            None => correspondence.violation(format!(
                                "stream #{index}: attempt names {effect}, which is not in the \
                                 `operation:step:N:effect:M` vocabulary"
                            )),
                            Some((operation, step)) => match segments.get(operation) {
                                None => correspondence.degraded(format!(
                                    "stream #{index}: attempt names {effect}, but operation \
                                     {operation} has no journal segment (the journal may cover \
                                     a subset of the session)"
                                )),
                                Some(outcome) => match &outcome.effects {
                                    None => correspondence.degraded(format!(
                                        "stream #{index}: segment {operation} could not be \
                                         re-planned, so {effect}'s publication is unverifiable"
                                    )),
                                    Some(effects) if effects.published.contains(effect) => {
                                        correspondence.checked += 1;
                                    }
                                    Some(effects) if step > effects.max_step => {
                                        correspondence.degraded(format!(
                                            "stream #{index}: attempt names {effect} at step \
                                             {step}, past the journal's tip (step {}) — a \
                                             prefix cannot disprove it",
                                            effects.max_step
                                        ));
                                    }
                                    Some(_) => correspondence.violation(format!(
                                        "stream #{index}: attempt names {effect}, but the \
                                         deterministic re-plan of {operation} never published \
                                         it — the attempt is unmoored from the journal"
                                    )),
                                },
                            },
                        },
                    }

                    // C6.2
                    match request_fingerprint.as_deref() {
                        None => fingerprints.violation(format!(
                            "stream #{index}: provider_attempt {label} without \
                             request_fingerprint — the writers require it (G2)"
                        )),
                        Some(fingerprint) if measured.contains(fingerprint) => {
                            fingerprints.checked += 1;
                        }
                        Some(fingerprint) => fingerprints.violation(format!(
                            "stream #{index}: attempt {label} carries fingerprint \
                             {fingerprint}, but no prompt_measured in this session carries it"
                        )),
                    }

                    // C6.3
                    match (route_id.as_deref(), baseline) {
                        (Some(route), Some(pinned)) if route != pinned => routes.violation(
                            format!(
                                "stream #{index}: attempt {label} ran on route {route} inside \
                                 a run pinned to {pinned} — an in-run route change is a \
                                 violation (Q3)"
                            ),
                        ),
                        (Some(_), Some(_)) => routes.checked += 1,
                        // No pinned run, or the attempt lacks a route: unverifiable.
                        _ => routes.unverifiable += 1,
                    }
                }
                _ => {}
            }
        }
    }

    vec![
        correspondence.report(
            "C6.1",
            total_attempts,
            "attempt↔journal effect correspondence",
        ),
        fingerprints.report("C6.2", total_attempts, "attempt fingerprint↔prompt_measured join"),
        routes.report("C6.3", total_attempts, "in-run route stability"),
    ]
}

/// One C6 clause's tally across every stream. Fail outranks degrade; a clause that found no
/// attempts at all degrades (a pre-0.2.63 log carries none — C7).
#[derive(Default)]
struct ClauseAccumulator {
    checked: usize,
    unverifiable: usize,
    violations: Vec<String>,
    degraded_notes: Vec<String>,
}

impl ClauseAccumulator {
    fn violation(&mut self, detail: String) {
        self.violations.push(detail);
    }

    fn degraded(&mut self, note: String) {
        self.degraded_notes.push(note);
    }

    fn report(self, rule: &str, total_attempts: usize, what: &str) -> RuleReport {
        let rule = rule.to_string();
        if !self.violations.is_empty() {
            return RuleReport {
                rule,
                verdict: Verdict::Fail,
                detail: self.violations.join("; "),
            };
        }
        if total_attempts == 0 {
            return RuleReport {
                rule,
                verdict: Verdict::Degraded,
                detail: format!(
                    "no provider_attempt events in any stream — a pre-0.2.63 log carries none \
                     (C7); {what} unchecked"
                ),
            };
        }
        if !self.degraded_notes.is_empty() || self.unverifiable > 0 {
            let mut detail = self.degraded_notes.join("; ");
            if self.unverifiable > 0 {
                if !detail.is_empty() {
                    detail.push_str("; ");
                }
                detail.push_str(&format!(
                    "{} attempt(s) unverifiable (no pinned run route)",
                    self.unverifiable
                ));
            }
            return RuleReport {
                rule,
                verdict: Verdict::Degraded,
                detail: format!(
                    "{} attempt(s) verified for {what}; {detail}",
                    self.checked
                ),
            };
        }
        RuleReport {
            rule,
            verdict: Verdict::Pass,
            detail: format!(
                "{} attempt(s) verified — {what} holds across every stream",
                self.checked
            ),
        }
    }
}

// ---------------------------------------------------------------------------------------------
// C8 · invocation chain adjacency (batch 3, 0.2.64 S4c)
// ---------------------------------------------------------------------------------------------

/// C8 · a retried invocation's chain must be journal-real. The SessionLog's falsifiable claim
/// is the endpoint pair: `llm_completed.invocation_id` (the chain's first effect — the derived
/// identity, P4 §1.1) and `llm_completed.effect_id` (the effect the kernel adopted). When they
/// differ, the journal must show that the head did NOT close the invocation:
///
/// - both ids parse in the `{operation}:step:N:effect:M` vocabulary, same operation (causation
///   cannot cross operations — C4's spirit), and the selected effect's step strictly follows
///   the head's;
/// - the head's resolution is **chain-advancing** — `Overflow` (the compaction ladder
///   republishes call_provider) or `Failed` (accepted for forward-compat: today's kernel
///   answers a CallProvider failure with a terminal per DEC-5, and a restored segment has
///   already proven the kernel itself walked whatever followed). A `Completed` head with a
///   *different* selected effect is the "merge two invocations into one" forgery: it fails.
///
/// Two deliberate deviations from P4 §5's letter, both forced by the wire reality:
/// 1. §5 says the hop between adjacent effects is a *Failed* resolution. Today's kernel never
///    re-emits after a CallProvider failure (DEC-5: `plan_effect_failure` → terminal), so real
///    chains advance through **Succeeded/ContextOverflow** resolutions. C8 checks
///    chain-advancing, not Failed, or every honest 0.2.63 overflow-retry log would read forged.
/// 2. §5's per-adjacent-pair walk needs published-effect causation, which the record format
///    deliberately omits (the step payload stays out of records — only step_digest). C8
///    verifies the chain's endpoints journal-directly and delegates the middle to the C3
///    re-plan's determinism: a segment that restored cleanly contains only steps the kernel's
///    own rules produced.
///
/// Plane lag degrades, never fails: the journal may trail the SessionLog, so a head whose
/// resolution hasn't landed yet (pending) or an effect claiming a step past the journal's tip
/// is unverifiable, not forged. A first-try chain (`invocation_id == effect_id`) has no
/// adjacency to prove. Old logs without `invocation_id` degrade per C7.
fn check_c8(streams: &[SessionStream], outcomes: &[SegmentOutcome]) -> RuleReport {
    let rule = "C8".to_string();
    let segments: HashMap<&str, &SegmentOutcome> = outcomes
        .iter()
        .filter(|outcome| outcome.report.operation_id != UNATTRIBUTED_SEGMENT)
        .map(|outcome| (outcome.report.operation_id.as_str(), outcome))
        .collect();

    let mut llm_completed_events = 0usize;
    let mut checked = 0usize;
    let mut trivial = 0usize;
    let mut unverifiable = 0usize;
    let mut violations: Vec<String> = Vec::new();
    let mut degraded_notes: Vec<String> = Vec::new();

    for (index, stream) in streams.iter().enumerate() {
        for event in &stream.events {
            let EvidenceEvent::LlmCompleted {
                effect_id,
                invocation_id,
            } = event
            else {
                continue;
            };
            llm_completed_events += 1;
            let (Some(head), Some(selected)) = (invocation_id.as_deref(), effect_id.as_deref())
            else {
                // A 0.2.62 log's llm_completed lacks the additive fields. C7: unverifiable,
                // never failed.
                unverifiable += 1;
                continue;
            };
            if head == selected {
                trivial += 1;
                continue;
            }

            let (Some((head_op, head_step)), Some((selected_op, selected_step))) =
                (parse_effect_step(head), parse_effect_step(selected))
            else {
                violations.push(format!(
                    "stream #{index}: llm_completed claims invocation {head} → {selected}, but \
                     one of the pair is not in the `operation:step:N:effect:M` vocabulary"
                ));
                continue;
            };
            if head_op != selected_op {
                violations.push(format!(
                    "stream #{index}: llm_completed claims invocation {head} → {selected} — an \
                     invocation chain cannot cross operations"
                ));
                continue;
            }
            if selected_step <= head_step {
                violations.push(format!(
                    "stream #{index}: llm_completed claims invocation {head} → {selected}, but \
                     the selected effect does not follow the chain head"
                ));
                continue;
            }
            let Some(outcome) = segments.get(head_op) else {
                degraded_notes.push(format!(
                    "stream #{index}: operation {head_op} has no journal segment (the journal \
                     may cover a subset of the session)"
                ));
                continue;
            };
            let Some(effects) = &outcome.effects else {
                degraded_notes.push(format!(
                    "stream #{index}: segment {head_op} could not be re-planned, so the \
                     invocation {head} → {selected} is unverifiable"
                ));
                continue;
            };

            // The head must be chain-advancing.
            match effects.resolutions.get(head) {
                Some(ResolutionFact::Completed) | Some(ResolutionFact::Other) => {
                    violations.push(format!(
                        "stream #{index}: llm_completed claims invocation {head} → {selected}, \
                         but {head} resolved to completion — a completed effect closes its \
                         invocation; nothing chains from it"
                    ));
                    continue;
                }
                Some(ResolutionFact::Overflow) | Some(ResolutionFact::Failed) => {}
                None if effects.published.contains(head) => degraded_notes.push(format!(
                    "stream #{index}: chain head {head} is published but its resolution has \
                     not landed in the journal (the planes are not synchronised)"
                )),
                None if head_step > effects.max_step => degraded_notes.push(format!(
                    "stream #{index}: chain head {head} claims step {head_step}, past the \
                     journal's tip (step {})",
                    effects.max_step
                )),
                None => {
                    violations.push(format!(
                        "stream #{index}: llm_completed claims invocation head {head}, but the \
                         deterministic re-plan of {head_op} never published it"
                    ));
                    continue;
                }
            }

            // The selected effect must exist on the chain.
            if effects.resolutions.contains_key(selected) {
                checked += 1;
            } else if effects.published.contains(selected) || selected_step > effects.max_step {
                degraded_notes.push(format!(
                    "stream #{index}: selected effect {selected} is not resolved in the \
                     journal (the planes are not synchronised)"
                ));
            } else {
                violations.push(format!(
                    "stream #{index}: llm_completed selects {selected}, but the deterministic \
                     re-plan of {selected_op} never published it"
                ));
            }
        }
    }

    if !violations.is_empty() {
        return RuleReport {
            rule,
            verdict: Verdict::Fail,
            detail: violations.join("; "),
        };
    }
    if llm_completed_events == 0 {
        return RuleReport {
            rule,
            verdict: Verdict::Degraded,
            detail: "no llm_completed events in any stream — invocation adjacency unchecked"
                .to_string(),
        };
    }
    if !degraded_notes.is_empty() || unverifiable > 0 {
        let mut detail = degraded_notes.join("; ");
        if unverifiable > 0 {
            if !detail.is_empty() {
                detail.push_str("; ");
            }
            detail.push_str(&format!(
                "{unverifiable} llm_completed event(s) without invocation_id/effect_id \
                 (pre-0.2.63 fields — C7)"
            ));
        }
        if !detail.is_empty() {
            return RuleReport {
                rule,
                verdict: Verdict::Degraded,
                detail: format!(
                    "{checked} retried invocation(s) verified, {trivial} first-try chain(s) \
                     closed; {detail}"
                ),
            };
        }
    }
    RuleReport {
        rule,
        verdict: Verdict::Pass,
        detail: format!(
            "{checked} retried invocation(s) verified end-to-end, {trivial} first-try chain(s) \
             closed"
        ),
    }
}

/// Strict first, lenient second: a record that fails the strict decode but still shows its
/// identity fields retains its context for C7 reporting. Proven digest corruption still fails
/// C1; only unavailable evidence degrades. Anything else is not a record.
fn classify(ordinal: usize, bytes: &[u8]) -> Option<Hop> {
    let error = match KernelRecord::from_record_bytes(bytes) {
        Ok(record) => return Some(Hop::Complete(record)),
        Err(error) => error,
    };
    let value: serde_json::Value = serde_json::from_slice(bytes).ok()?;
    let object = value.as_object()?;
    let string = |key: &str| {
        object
            .get(key)
            .and_then(serde_json::Value::as_str)
            .map(str::to_string)
    };
    // `step_seq` rides the wire as a branded decimal string (scalar.rs), but an old-format or
    // foreign record may carry a bare number — accept both.
    let step_seq = object.get("step_seq").and_then(|value| {
        value
            .as_u64()
            .or_else(|| value.as_str().and_then(|text| text.parse().ok()))
    });
    let degraded = DegradedRecord {
        ordinal,
        operation_id: string("operation_id"),
        input_id: string("input_id"),
        step_seq,
        previous_record_digest: string("previous_record_digest"),
        record_digest: string("record_digest"),
        reason: format!("{}: {}", error.code().as_str(), error.message()),
        integrity_failure: matches!(error, RecordError::DigestMismatch(_)),
    };
    // An old-format record must still answer "which chain, which hop" to count as evidence;
    // without either it is unparseable input.
    if degraded.operation_id.is_some() || degraded.step_seq.is_some() {
        Some(Hop::Degraded(degraded))
    } else {
        None
    }
}

fn validate_segment(operation_id: &str, mut hops: Vec<Hop>) -> SegmentOutcome {
    // The chain's own fields define the order; the input order is a storage detail. Hops that
    // cannot say where they sit sort last, in input order.
    hops.sort_by_key(|hop| {
        (
            hop.step_seq().unwrap_or(u64::MAX),
            match hop {
                Hop::Complete(_) => 0usize,
                Hop::Degraded(degraded) => degraded.ordinal,
            },
        )
    });

    let degraded_hops: Vec<DegradedHop> = hops
        .iter()
        .filter_map(|hop| match hop {
            Hop::Degraded(degraded) => Some(degraded.marking()),
            Hop::Complete(_) => None,
        })
        .collect();
    let hop_count = hops.len();

    let c1 = check_c1(&hops);
    let replan = replan_segment(&hops, &c1);
    let c2 = check_c2(&hops);
    let c3 = render_c3(&replan);
    let c4 = check_c4(&hops, operation_id);

    // C6.1's membership evidence: when the re-plan ran, the operation's published effects are
    // exactly (journal-resolved effect ids) ∪ (still-pending effects after the re-plan).
    let effects = match &replan {
        Replan::Restored(restored) => {
            let mut published: std::collections::HashSet<String> = std::collections::HashSet::new();
            let mut resolutions: HashMap<String, ResolutionFact> = HashMap::new();
            for hop in &hops {
                if let Some((effect_id, fact)) = resolution_of(hop) {
                    published.insert(effect_id.clone());
                    resolutions.insert(effect_id, fact);
                }
            }
            published.extend(
                restored
                    .transaction
                    .pending_effects()
                    .map(|effect| effect.effect_id.as_str().to_string()),
            );
            let max_step = hops.iter().filter_map(|hop| hop.step_seq()).max().unwrap_or(0);
            Some(SegmentEffects {
                max_step,
                published,
                resolutions,
            })
        }
        _ => None,
    };

    SegmentOutcome {
        report: SegmentReport {
            operation_id: operation_id.to_string(),
            hops: hop_count,
            degraded_hops,
            rules: vec![c1, c2, c3, c4],
        },
        effects,
    }
}

/// A segment's verdict plus the cross-plane evidence C6 needs from it.
struct SegmentOutcome {
    report: SegmentReport,
    /// `Some` iff the deterministic re-plan ran (the same condition under which C3 passes).
    effects: Option<SegmentEffects>,
}

struct SegmentEffects {
    /// The highest step the journal reaches — an attempt naming a later step is unverifiable
    /// (prefix), not forged.
    max_step: u64,
    /// Every effect the operation published through the journal's tip.
    published: std::collections::HashSet<String>,
    /// The journal-direct resolution fact per resolved effect — C8's adjacency evidence.
    resolutions: HashMap<String, ResolutionFact>,
}

/// How a resolved effect's outcome bears on an invocation chain (C8). The two planes are not
/// synchronised, so this is read only on fully restored segments.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ResolutionFact {
    /// Succeeded with `ProviderOutcome::Completed` — closes the invocation; nothing chains.
    Completed,
    /// Succeeded with `ProviderOutcome::ContextOverflow` — the compaction ladder republishes
    /// call_provider: the one chain-advancing resolution in today's kernel (see check_c8).
    Overflow,
    /// `EffectOutcome::Failed` — terminal for CallProvider under DEC-5 today, but
    /// chain-advancing under any future kernel with a failure retry ladder; a restored
    /// segment has already proven the kernel itself walked whatever follows.
    Failed,
    /// Any other resolution (tools/spawn/syscall/...) — never invocation-chain-advancing.
    Other,
}

/// The effect a record's ResolveEffect input settles and how, if it is one — the
/// journal-direct resolution facts (the same read C4 makes, with the outcome kept).
fn resolution_of(hop: &Hop) -> Option<(String, ResolutionFact)> {
    let Hop::Complete(record) = hop else { return None };
    let input = record.normalized_input().ok()?;
    let NormalizedPayload::ResolveEffect(resolve) = &input.input else {
        return None;
    };
    let fact = match &resolve.outcome {
        EffectOutcome::Failed(_) => ResolutionFact::Failed,
        EffectOutcome::Succeeded(success) => match &success.result {
            EffectSuccess::Provider(provider) => match &provider.outcome {
                ProviderOutcome::Completed(_) => ResolutionFact::Completed,
                ProviderOutcome::ContextOverflow(_) => ResolutionFact::Overflow,
            },
            _ => ResolutionFact::Other,
        },
    };
    Some((resolve.effect_id.as_str().to_string(), fact))
}

/// C1 · chain integrity.
fn check_c1(hops: &[Hop]) -> RuleReport {
    let rule = "C1".to_string();
    if hops.is_empty() {
        return RuleReport {
            rule,
            verdict: Verdict::Degraded,
            detail: "no records in this segment".to_string(),
        };
    }
    let all_complete = hops.iter().all(|hop| matches!(hop, Hop::Complete(_)));
    if all_complete {
        let records: Vec<KernelRecord> = hops
            .iter()
            .filter_map(|hop| match hop {
                Hop::Complete(record) => Some(record.clone()),
                Hop::Degraded(_) => None,
            })
            .collect();
        return match verify_record_chain(&records) {
            Ok(genesis_digest) => RuleReport {
                rule,
                verdict: Verdict::Pass,
                detail: format!(
                    "{} record(s), genesis {genesis_digest}, every link verified",
                    records.len()
                ),
            },
            Err(error) => RuleReport {
                rule,
                verdict: Verdict::Fail,
                detail: format!("{}: {}", error.code().as_str(), error.message()),
            },
        };
    }

    // Mixed segment: check every link whose digests survived, and the genesis claim when the
    // first hop can make one. Degraded hops verify nothing themselves.
    let mut broken: Vec<String> = hops
        .iter()
        .filter_map(|hop| match hop {
            Hop::Degraded(record) if record.integrity_failure => Some(record.reason.clone()),
            _ => None,
        })
        .collect();
    let mut unverifiable_links = 0usize;
    let mut previous: Option<(&Hop, Option<&KernelRecord>)> = None;
    for hop in hops {
        let step = hop.step_seq();
        let (prev_digest, _) = digests_of(hop);
        if let Some((previous_hop, previous_complete)) = previous {
            let previous_step = previous_hop.step_seq();
            let previous_digest = digests_of(previous_hop).1;
            match (prev_digest, previous_digest) {
                (Some(expected), Some(actual)) if expected != actual => broken.push(format!(
                    "hop at step {} expects head {expected}, but its predecessor's digest is \
                     {actual}",
                    step.map_or("?".to_string(), |seq| seq.to_string()),
                )),
                (None, _) => unverifiable_links += 1,
                (_, None) => unverifiable_links += 1,
                _ => {}
            }
            match (step, previous_step) {
                (Some(step), Some(previous_step)) if step != previous_step + 1 => broken.push(
                    format!("hop is step {step}, but its predecessor is step {previous_step}"),
                ),
                (Some(_), Some(_)) => {}
                _ => unverifiable_links += 1,
            }
            // `verify_follows` is only meaningful across an unbroken run of complete records:
            // a degraded hop in between severs the chain of custody for the +1/digest pair.
            if let (Hop::Complete(record), Some(previous_record)) = (hop, previous_complete)
                && let Err(error) = record.verify_follows(Some(previous_record))
            {
                broken.push(format!("{}: {}", error.code().as_str(), error.message()));
            }
        } else if let Hop::Complete(record) = hop
            && let Err(error) = record.verify_follows(None)
        {
            broken.push(format!("{}: {}", error.code().as_str(), error.message()));
        }
        previous = Some((
            hop,
            match hop {
                Hop::Complete(record) => Some(record),
                Hop::Degraded(_) => None,
            },
        ));
    }

    if !broken.is_empty() {
        return RuleReport {
            rule,
            verdict: Verdict::Fail,
            detail: broken.join("; "),
        };
    }
    RuleReport {
        rule,
        verdict: Verdict::Degraded,
        detail: format!(
            "partial chain: every surviving link verified, {unverifiable_links} link(s) \
             unverifiable across degraded hop(s)"
        ),
    }
}

/// C2 · input idempotency: one input_id, one record.
fn check_c2(hops: &[Hop]) -> RuleReport {
    let rule = "C2".to_string();
    let mut by_input: HashMap<&str, &str> = HashMap::new();
    let mut conflicts: Vec<String> = Vec::new();
    let mut retries = 0usize;
    let mut unverifiable = 0usize;
    for hop in hops {
        let (input_id, record_digest) = match hop {
            Hop::Complete(record) => (
                Some(record.input_id().as_str()),
                Some(record.record_digest().as_str()),
            ),
            Hop::Degraded(degraded) => (
                degraded.input_id.as_deref(),
                degraded.record_digest.as_deref(),
            ),
        };
        let Some(input_id) = input_id else { continue };
        let Some(digest) = record_digest else {
            unverifiable += 1;
            continue;
        };
        match by_input.get(input_id) {
            Some(existing) if *existing != digest => conflicts.push(format!(
                "input {input_id} produced two different records ({existing} and {digest}); a \
                 retry must reach the same record"
            )),
            Some(_) => retries += 1,
            None => {
                by_input.insert(input_id, digest);
            }
        }
    }
    if !conflicts.is_empty() {
        return RuleReport {
            rule,
            verdict: Verdict::Fail,
            detail: conflicts.join("; "),
        };
    }
    if unverifiable > 0 {
        return RuleReport {
            rule,
            verdict: Verdict::Degraded,
            detail: format!(
                "{} unique input(s), {retries} idempotent retry hit(s); {unverifiable} degraded \
                 hop(s) could not be compared",
                by_input.len(),
            ),
        };
    }
    RuleReport {
        rule,
        verdict: Verdict::Pass,
        detail: format!(
            "{} unique input(s), {retries} idempotent retry hit(s), no divergent duplicates",
            by_input.len(),
        ),
    }
}

/// C3 · causal closure runs on the §12.2 genesis-leg restore: the deterministic re-plan of
/// every transition. Batch 3 shares that one restore with C6.1 — the restored transaction
/// answers "did this operation ever publish effect X" — so the restore happens once per
/// segment, here, and C3's report only renders the outcome.
enum Replan {
    /// C3/C6.1 degrade: the re-plan never ran.
    Unavailable(&'static str),
    /// The restore itself faulted — C3 fails.
    Failed(String),
    Restored(RestoredOperation),
}

fn replan_segment(hops: &[Hop], c1: &RuleReport) -> Replan {
    if hops.iter().any(|hop| matches!(hop, Hop::Degraded(_))) {
        return Replan::Unavailable(
            "re-plan requires complete records; this segment has degraded hops",
        );
    }
    if c1.verdict == Verdict::Fail {
        return Replan::Unavailable(
            "C1 failed; a re-plan over a broken chain would only re-report that break",
        );
    }
    let records: Vec<KernelRecord> = hops
        .iter()
        .filter_map(|hop| match hop {
            Hop::Complete(record) => Some(record.clone()),
            Hop::Degraded(_) => None,
        })
        .collect();
    if records.is_empty() {
        return Replan::Unavailable("no records in this segment");
    }
    match restore_operation(
        None,
        &records,
        ConfigDefaults::default(),
        InMemoryRecordIndex::from_records(&records),
    ) {
        Ok(restored) => Replan::Restored(restored),
        Err(fault) => Replan::Failed(format!("{}: {}", fault.code.as_str(), fault.message)),
    }
}

fn render_c3(replan: &Replan) -> RuleReport {
    let rule = "C3".to_string();
    match replan {
        Replan::Unavailable(reason) => RuleReport {
            rule,
            verdict: Verdict::Degraded,
            detail: (*reason).to_string(),
        },
        Replan::Failed(fault) => RuleReport {
            rule,
            verdict: Verdict::Fail,
            detail: fault.clone(),
        },
        Replan::Restored(restored) => RuleReport {
            rule,
            verdict: Verdict::Pass,
            detail: format!(
                "re-planned {} record(s) from genesis; every durable record digest reproduced",
                restored.cost.records_before_checkpoint
            ),
        },
    }
}

/// C4 · task lineage, the journal-direct half.
fn check_c4(hops: &[Hop], operation_id: &str) -> RuleReport {
    let rule = "C4".to_string();
    struct LaunchFact {
        task_id: String,
        attempt_id: String,
        step_seq: u64,
        effect_id: String,
    }

    let mut launches: Vec<LaunchFact> = Vec::new();
    let mut unreadable_inputs = 0usize;
    for hop in hops {
        let Hop::Complete(record) = hop else { continue };
        let input = match record.normalized_input() {
            Ok(input) => input,
            Err(_) => {
                unreadable_inputs += 1;
                continue;
            }
        };
        let NormalizedPayload::ResolveEffect(resolve) = &input.input else {
            continue;
        };
        let EffectOutcome::Succeeded(success) = &resolve.outcome else {
            continue;
        };
        let EffectSuccess::TasksSpawned(spawned) = &success.result else {
            continue;
        };
        for attempt in &spawned.attempts {
            launches.push(LaunchFact {
                task_id: attempt.task_id.as_str().to_string(),
                attempt_id: attempt.attempt_id.as_str().to_string(),
                step_seq: record.step_seq().get(),
                effect_id: resolve.effect_id.as_str().to_string(),
            });
        }
    }

    let mut violations: Vec<String> = Vec::new();
    let mut seen: HashMap<(&str, &str), u64> = HashMap::new();
    for fact in &launches {
        let pair = (fact.task_id.as_str(), fact.attempt_id.as_str());
        if let Some(first_step) = seen.insert(pair, fact.step_seq) {
            violations.push(format!(
                "task {} attempt {} launched at steps {first_step} and {}; the launch token is \
                 derived from that pair, so a repeated pair is a reused LaunchToken",
                fact.task_id, fact.attempt_id, fact.step_seq,
            ));
        }
        match parse_effect_step(&fact.effect_id) {
            Some((effect_operation, effect_step)) => {
                if effect_operation != operation_id {
                    violations.push(format!(
                        "task {} launch at step {} resolves effect {} of another operation — \
                         causation cannot cross operations",
                        fact.task_id, fact.step_seq, fact.effect_id,
                    ));
                } else if effect_step >= fact.step_seq {
                    violations.push(format!(
                        "task {} launch resolved at step {} names an effect published at step \
                         {effect_step} — the resolution precedes the publication",
                        fact.task_id, fact.step_seq,
                    ));
                }
            }
            None => violations.push(format!(
                "task {} launch at step {} names effect {}, which is not in the \
                 `operation:step:N:effect:M` vocabulary",
                fact.task_id, fact.step_seq, fact.effect_id,
            )),
        }
    }

    if !violations.is_empty() {
        return RuleReport {
            rule,
            verdict: Verdict::Fail,
            detail: violations.join("; "),
        };
    }
    let degraded_hops = hops
        .iter()
        .filter(|hop| matches!(hop, Hop::Degraded(_)))
        .count();
    if degraded_hops > 0 || unreadable_inputs > 0 {
        return RuleReport {
            rule,
            verdict: Verdict::Degraded,
            detail: format!(
                "{} launch(es) checked; {degraded_hops} degraded hop(s) and \
                 {unreadable_inputs} unreadable input(s) could hide further launches",
                launches.len(),
            ),
        };
    }
    RuleReport {
        rule,
        verdict: Verdict::Pass,
        detail: format!(
            "{} launch(es), every (task_id, attempt_id) pair unique, every spawn resolution \
             names an earlier step of this operation",
            launches.len(),
        ),
    }
}

/// The kernel's effect-id vocabulary is `{operation}:step:{N}:effect:{M}` (driver minting).
/// Operation ids may themselves contain colons, so parse from the right.
fn parse_effect_step(effect_id: &str) -> Option<(&str, u64)> {
    let (before_effect, _) = effect_id.rsplit_once(":effect:")?;
    let (operation, step) = before_effect.rsplit_once(":step:")?;
    Some((operation, step.parse().ok()?))
}

fn digests_of(hop: &Hop) -> (Option<&str>, Option<&str>) {
    match hop {
        Hop::Complete(record) => (
            record
                .previous_record_digest()
                .map(|digest| digest.as_str()),
            Some(record.record_digest().as_str()),
        ),
        Hop::Degraded(degraded) => (
            degraded.previous_record_digest.as_deref(),
            degraded.record_digest.as_deref(),
        ),
    }
}

// ---------------------------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use serde_json::json;

    use super::*;
    use crate::runtime::kernel::wire::config::{
        ConfigDefaults, ExecutionPolicy, HostEffectSupport, OperationConfig,
    };
    use crate::runtime::kernel::wire::driver::CanonicalOperationDriver;
    use crate::runtime::kernel::wire::effect::{
        EffectKindTag, EffectSucceeded, ProviderCompleted, ProviderContextOverflow,
        ProviderMessage, ProviderOutcome, ProviderSuccess, TaskLaunchOutcome, TaskLaunchStarted,
        TaskLaunchStatus, TasksSpawnedSuccess, ToolCall,
    };
    use crate::runtime::kernel::wire::envelope::{
        ConfigureOperation, KernelInput, ResolveEffect, StartOperation, WireEnvelope,
    };
    use crate::runtime::kernel::wire::record::{KernelRecord, NormalizedInput};
    use crate::runtime::kernel::wire::root::{
        InitialContext, LogicalAgentSpec, LogicalMessage, LogicalTask, MessageRole,
        RootAgentEntry, RootEntry, RootWorkflowEntry, WorkflowNode, WorkflowSpec,
    };
    use crate::runtime::kernel::wire::scalar::{
        AttemptId, BoundedJson, CallId, EffectId, InputId, NodeId, OperationId, TaskId, WireU64,
    };
    use crate::runtime::kernel::wire::transaction::{InMemoryRecordIndex, KernelTransaction};

    // -----------------------------------------------------------------------------------------
    // envelopes
    // -----------------------------------------------------------------------------------------

    fn operation(id: &str) -> OperationId {
        OperationId::new(id).unwrap()
    }

    fn envelope(op: &OperationId, id: &str, at: u64, input: KernelInput) -> WireEnvelope {
        WireEnvelope::new(
            op.clone(),
            InputId::new(id).unwrap(),
            WireU64::new(at),
            input,
        )
    }

    fn configure_envelope(op: &OperationId) -> WireEnvelope {
        envelope(
            op,
            "in-configure",
            1_700_000_000_000,
            KernelInput::ConfigureOperation(ConfigureOperation {
                config: OperationConfig {
                    execution_policy: Some(ExecutionPolicy {
                        max_turns: Some(12),
                        ..ExecutionPolicy::default()
                    }),
                    host_effect_support: HostEffectSupport::new([
                        EffectKindTag::CallProvider,
                        EffectKindTag::SpawnTasks,
                    ]),
                    ..OperationConfig::default()
                },
            }),
        )
    }

    fn agent_start_envelope(op: &OperationId) -> WireEnvelope {
        envelope(
            op,
            "in-start",
            1_700_000_001_000,
            KernelInput::StartOperation(StartOperation {
                entry: RootEntry::Agent(RootAgentEntry {
                    task: LogicalTask::new("write the brief"),
                    run_spec: Some(LogicalAgentSpec::new("write the brief")),
                }),
                initial_context: InitialContext::default(),
            }),
        )
    }

    /// An agent start carrying `messages` history items — the compaction ladder needs real
    /// history to reclaim, or the first context overflow exhausts recovery and terminates the
    /// operation instead of republishing a call_provider effect.
    fn agent_start_with_history_envelope(op: &OperationId, messages: usize) -> WireEnvelope {
        envelope(
            op,
            "in-start",
            1_700_000_001_000,
            KernelInput::StartOperation(StartOperation {
                entry: RootEntry::Agent(RootAgentEntry {
                    task: LogicalTask::new("write the brief"),
                    run_spec: Some(LogicalAgentSpec::new("write the brief")),
                }),
                initial_context: InitialContext {
                    messages: (0..messages)
                        .map(|index| LogicalMessage {
                            role: if index % 2 == 0 {
                                MessageRole::User
                            } else {
                                MessageRole::Assistant
                            },
                            content: format!(
                                "turn {index}: a long enough body that compaction has \
                                 something to reclaim when the prompt stops fitting"
                            ),
                            tokens: Some(64),
                            tool_call_id: None,
                        })
                        .collect(),
                    ..InitialContext::default()
                },
            }),
        )
    }

    fn workflow_start_envelope(op: &OperationId) -> WireEnvelope {
        envelope(
            op,
            "in-start",
            1_700_000_001_000,
            KernelInput::StartOperation(StartOperation {
                entry: RootEntry::Workflow(RootWorkflowEntry {
                    spec: WorkflowSpec {
                        name: "brief".to_string(),
                        nodes: vec![
                            WorkflowNode {
                                node_id: NodeId::new("collect").unwrap(),
                                task: LogicalTask::new("collect the sources"),
                                depends_on: vec![],
                                run_spec: Some(LogicalAgentSpec::new("collect the sources")),
                            },
                            WorkflowNode {
                                node_id: NodeId::new("write").unwrap(),
                                task: LogicalTask::new("write the brief"),
                                depends_on: vec![NodeId::new("collect").unwrap()],
                                run_spec: Some(LogicalAgentSpec::new("write the brief")),
                            },
                        ],
                    },
                }),
                initial_context: InitialContext::default(),
            }),
        )
    }

    fn resolve_overflow_envelope(op: &OperationId, effect_step: u64) -> WireEnvelope {
        envelope(
            op,
            "in-resolve",
            1_700_000_002_000,
            KernelInput::ResolveEffect(ResolveEffect {
                effect_id: EffectId::new(format!("{op}:step:{effect_step}:effect:0")).unwrap(),
                outcome: EffectOutcome::Succeeded(EffectSucceeded {
                    result: EffectSuccess::Provider(ProviderSuccess {
                        outcome: ProviderOutcome::ContextOverflow(
                            ProviderContextOverflow::default(),
                        ),
                    }),
                }),
            }),
        )
    }

    /// A provider completion. `with_tool_call` makes the completion request a tool, so the
    /// next step publishes an ExecuteTools effect instead of terminating the operation.
    fn resolve_completed_envelope(
        op: &OperationId,
        id: &str,
        at: u64,
        effect_step: u64,
        with_tool_call: bool,
    ) -> WireEnvelope {
        envelope(
            op,
            id,
            at,
            KernelInput::ResolveEffect(ResolveEffect {
                effect_id: EffectId::new(format!("{op}:step:{effect_step}:effect:0")).unwrap(),
                outcome: EffectOutcome::Succeeded(EffectSucceeded {
                    result: EffectSuccess::Provider(ProviderSuccess {
                        outcome: ProviderOutcome::Completed(ProviderCompleted {
                            message: ProviderMessage {
                                role: MessageRole::Assistant,
                                content: "done".to_string(),
                                tool_calls: if with_tool_call {
                                    vec![ToolCall {
                                        call_id: CallId::new("call-1").unwrap(),
                                        name: "read_file".to_string(),
                                        arguments: BoundedJson::new(json!({})).unwrap(),
                                    }]
                                } else {
                                    Vec::new()
                                },
                                tool_call_id: None,
                                tokens: None,
                            },
                            observed_input_tokens: None,
                            observed_output_tokens: None,
                            stop_reason: None,
                        }),
                    }),
                }),
            }),
        )
    }

    fn resolve_spawn_envelope(
        op: &OperationId,
        id: &str,
        at: u64,
        effect_id: &str,
        tasks: &[(&str, &str)],
    ) -> WireEnvelope {
        envelope(
            op,
            id,
            at,
            KernelInput::ResolveEffect(ResolveEffect {
                effect_id: EffectId::new(effect_id).unwrap(),
                outcome: EffectOutcome::Succeeded(EffectSucceeded {
                    result: EffectSuccess::TasksSpawned(TasksSpawnedSuccess {
                        attempts: tasks
                            .iter()
                            .map(|(task, attempt)| TaskLaunchOutcome {
                                task_id: TaskId::new(*task).unwrap(),
                                attempt_id: AttemptId::new(*attempt).unwrap(),
                                outcome: TaskLaunchStatus::Started(TaskLaunchStarted {}),
                            })
                            .collect(),
                    }),
                }),
            }),
        )
    }

    // -----------------------------------------------------------------------------------------
    // chain builders
    // -----------------------------------------------------------------------------------------

    /// The honest path: a live transaction driven by the real driver, so every record's step is
    /// exactly what a re-plan reproduces. This is what a host's journal prefix looks like.
    fn live_chain(envelopes: &[WireEnvelope]) -> Vec<KernelRecord> {
        let mut tx = KernelTransaction::new(ConfigDefaults::default(), InMemoryRecordIndex::new());
        let mut driver = CanonicalOperationDriver::new();
        let mut journal = Vec::new();
        for envelope in envelopes {
            let preparation = tx.prepare(envelope, |context| driver.plan(context));
            let token = preparation
                .token()
                .unwrap_or_else(|| {
                    panic!("expected a prepared step, got {:?}", preparation.fault())
                })
                .clone();
            let head = preparation.record().unwrap().record_digest().clone();
            let committed = tx.commit(&token, &head).expect("commit must succeed");
            journal.push(committed.record.clone());
            driver
                .note_committed(committed.step_seq)
                .expect("the driver folds the step it planned");
        }
        journal
    }

    /// A structurally sound chain whose steps are hand-pinned JSON — **not** the driver's plans.
    /// C1/C2/C4 read only the records, so they judge these chains; C3 necessarily fails on them
    /// (the re-plan cannot reproduce a hand-pinned step) and is simply not asserted there.
    fn hand_chain(envelopes: &[WireEnvelope]) -> Vec<KernelRecord> {
        let mut records: Vec<KernelRecord> = Vec::new();
        for (index, envelope) in envelopes.iter().enumerate() {
            let input = NormalizedInput::normalize(envelope, &ConfigDefaults::default())
                .expect("the envelope normalises");
            let step = json!({ "planned": format!("step-{index}"), "effects": [] });
            let record =
                KernelRecord::chain(records.last(), &input, &step).expect("the record chains");
            records.push(record);
        }
        records
    }

    fn blobs(records: &[KernelRecord]) -> Vec<Vec<u8>> {
        records
            .iter()
            .map(|record| record.record_bytes().into_vec())
            .collect()
    }

    fn rule<'a>(report: &'a ValidationReport, segment: usize, id: &str) -> &'a RuleReport {
        report.segments[segment]
            .rules
            .iter()
            .find(|rule| rule.rule == id)
            .unwrap_or_else(|| panic!("segment {segment} has no {id} verdict"))
    }

    fn cross<'a>(report: &'a ValidationReport, id: &str) -> &'a RuleReport {
        report
            .cross_checks
            .iter()
            .find(|rule| rule.rule == id)
            .unwrap_or_else(|| panic!("the report has no {id} cross-check"))
    }

    // -----------------------------------------------------------------------------------------
    // green paths
    // -----------------------------------------------------------------------------------------

    #[test]
    fn a_green_agent_chain_passes_every_rule() {
        let op = operation("op-green-agent");
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_overflow_envelope(&op, 1),
        ]);
        let report = validate_journal(&blobs(&chain));
        assert_eq!(report.segments.len(), 1);
        for id in ["C1", "C2", "C3", "C4"] {
            assert_eq!(
                rule(&report, 0, id).verdict,
                Verdict::Pass,
                "{id}: {}",
                rule(&report, 0, id).detail
            );
        }
        assert!(
            rule(&report, 0, "C3")
                .detail
                .contains("every durable record digest reproduced"),
            "C3 proves the re-plan: {}",
            rule(&report, 0, "C3").detail
        );
        assert_eq!(report.exit_code(), 0);
        assert_eq!(report.unparseable_records, 0);
    }

    #[test]
    fn a_green_workflow_chain_passes_c4_with_real_launches() {
        let op = operation("op-green-workflow");
        let chain = live_chain(&[
            configure_envelope(&op),
            workflow_start_envelope(&op),
            resolve_spawn_envelope(
                &op,
                "in-ack-1",
                1_700_000_002_000,
                "op-green-workflow:step:1:effect:0",
                &[("wf-node0", "wf-node0:attempt:1")],
            ),
        ]);
        let report = validate_journal(&blobs(&chain));
        assert_eq!(report.segments.len(), 1);
        for id in ["C1", "C2", "C3", "C4"] {
            assert_eq!(
                rule(&report, 0, id).verdict,
                Verdict::Pass,
                "{id}: {}",
                rule(&report, 0, id).detail
            );
        }
        assert!(
            rule(&report, 0, "C4").detail.contains("1 launch(es)"),
            "{}",
            rule(&report, 0, "C4").detail
        );
        assert_eq!(report.deferred.len(), 2, "batch-1 scope limits are named");
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn input_order_is_a_storage_detail() {
        let op = operation("op-shuffled");
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_overflow_envelope(&op, 1),
        ]);
        let mut shuffled = blobs(&chain);
        shuffled.reverse();
        let report = validate_journal(&shuffled);
        assert_eq!(
            report.exit_code(),
            0,
            "the chain's own links define the order"
        );
    }

    #[test]
    fn two_operations_validate_as_independent_segments() {
        let op_a = operation("op-seg-a");
        let op_b = operation("op-seg-b");
        let chain_a = live_chain(&[configure_envelope(&op_a), agent_start_envelope(&op_a)]);
        let chain_b = live_chain(&[configure_envelope(&op_b), agent_start_envelope(&op_b)]);
        // Interleaved and sharing input ids — idempotency is namespaced per operation.
        let mut mixed = Vec::new();
        for index in 0..2 {
            mixed.push(chain_a[index].record_bytes().into_vec());
            mixed.push(chain_b[index].record_bytes().into_vec());
        }
        let report = validate_journal(&mixed);
        assert_eq!(report.segments.len(), 2);
        assert_eq!(report.exit_code(), 0);
    }

    // -----------------------------------------------------------------------------------------
    // C1 · chain integrity
    // -----------------------------------------------------------------------------------------

    #[test]
    fn a_gap_in_the_chain_fails_c1_and_degrades_c3() {
        let op = operation("op-gapped");
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_overflow_envelope(&op, 1),
        ]);
        let gapped = blobs(&[chain[0].clone(), chain[2].clone()]);
        let report = validate_journal(&gapped);
        assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Fail);
        assert_eq!(
            rule(&report, 0, "C3").verdict,
            Verdict::Degraded,
            "a re-plan over a broken chain would only re-report the C1 break"
        );
        assert_eq!(report.exit_code(), 1);
    }

    // -----------------------------------------------------------------------------------------
    // C2 · input idempotency
    // -----------------------------------------------------------------------------------------

    #[test]
    fn two_different_records_for_one_input_fail_c2() {
        let op = operation("op-dup-input");
        // Two chains over the same operation id whose `in-start` envelopes differ only in the
        // observed clock — same input id, different canonical input, different records.
        let chain_a = hand_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let mut later_start = agent_start_envelope(&op);
        later_start.observed_at_ms = WireU64::new(1_700_000_001_500);
        let chain_b = hand_chain(&[configure_envelope(&op), later_start]);
        assert_ne!(
            chain_a[1].record_digest(),
            chain_b[1].record_digest(),
            "the fixture must produce two different records for one input id"
        );
        let report = validate_journal(&blobs(&[
            chain_a[0].clone(),
            chain_a[1].clone(),
            chain_b[1].clone(),
        ]));
        assert_eq!(rule(&report, 0, "C2").verdict, Verdict::Fail);
        assert!(
            rule(&report, 0, "C2").detail.contains("in-start"),
            "{}",
            rule(&report, 0, "C2").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    // -----------------------------------------------------------------------------------------
    // C4 · task lineage
    // -----------------------------------------------------------------------------------------

    #[test]
    fn a_repeated_attempt_pair_is_a_reused_launch_token() {
        let op = operation("op-dup-launch");
        let chain = hand_chain(&[
            configure_envelope(&op),
            resolve_spawn_envelope(
                &op,
                "in-ack-1",
                1_700_000_001_000,
                "op-dup-launch:step:0:effect:0",
                &[("writer", "writer:attempt:1")],
            ),
            resolve_spawn_envelope(
                &op,
                "in-ack-2",
                1_700_000_002_000,
                "op-dup-launch:step:0:effect:0",
                &[("writer", "writer:attempt:1")],
            ),
        ]);
        let report = validate_journal(&blobs(&chain));
        assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Pass);
        assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Fail);
        assert!(
            rule(&report, 0, "C4").detail.contains("LaunchToken"),
            "the verdict names the token reuse: {}",
            rule(&report, 0, "C4").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    #[test]
    fn a_resolution_naming_a_future_step_fails_c4() {
        let op = operation("op-future-effect");
        let chain = hand_chain(&[
            configure_envelope(&op),
            resolve_spawn_envelope(
                &op,
                "in-ack-1",
                1_700_000_001_000,
                "op-future-effect:step:5:effect:0",
                &[("writer", "writer:attempt:1")],
            ),
        ]);
        let report = validate_journal(&blobs(&chain));
        assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Fail);
        assert!(
            rule(&report, 0, "C4")
                .detail
                .contains("precedes the publication"),
            "{}",
            rule(&report, 0, "C4").detail
        );
    }

    #[test]
    fn a_resolution_naming_another_operation_fails_c4() {
        let op = operation("op-foreign-effect");
        let chain = hand_chain(&[
            configure_envelope(&op),
            resolve_spawn_envelope(
                &op,
                "in-ack-1",
                1_700_000_001_000,
                "op-somewhere-else:step:0:effect:0",
                &[("writer", "writer:attempt:1")],
            ),
        ]);
        let report = validate_journal(&blobs(&chain));
        assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Fail);
        assert!(
            rule(&report, 0, "C4").detail.contains("another operation"),
            "{}",
            rule(&report, 0, "C4").detail
        );
    }

    // -----------------------------------------------------------------------------------------
    // C7 · degradation
    // -----------------------------------------------------------------------------------------

    #[test]
    fn a_tampered_hop_fails_integrity_validation() {
        let op = operation("op-tampered");
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_overflow_envelope(&op, 1),
        ]);
        let mut input = blobs(&chain);
        // Corrupt the middle record's step_digest: the strict decode now fails the self-digest
        // check. Surviving identity fields must not hide proven corruption.
        let mut forged: serde_json::Value = serde_json::from_slice(&input[1]).unwrap();
        forged["step_digest"] = serde_json::Value::String(chain[0].record_digest().to_string());
        input[1] = serde_json::to_vec(&forged).unwrap();

        let report = validate_journal(&input);
        assert_eq!(report.segments.len(), 1);
        assert_eq!(report.segments[0].degraded_hops.len(), 1);
        assert_eq!(
            rule(&report, 0, "C1").verdict,
            Verdict::Fail,
            "a digest mismatch must fail C1: {}",
            rule(&report, 0, "C1").detail
        );
        assert_eq!(rule(&report, 0, "C3").verdict, Verdict::Degraded);
        assert_eq!(rule(&report, 0, "C4").verdict, Verdict::Degraded);
        assert_eq!(
            report.exit_code(),
            1,
            "proven digest corruption must fail the validator"
        );
    }

    #[test]
    fn missing_legacy_digest_degrades_without_claiming_corruption() {
        let op = operation("op-legacy");
        let chain = live_chain(&[configure_envelope(&op)]);
        let mut legacy: serde_json::Value = serde_json::from_slice(&blobs(&chain)[0]).unwrap();
        legacy.as_object_mut().unwrap().remove("step_digest");
        let report = validate_journal(&[serde_json::to_vec(&legacy).unwrap()]);
        assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Degraded);
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn unparseable_input_is_evidence_insufficient_not_guilty() {
        let report = validate_journal(&[b"this is not a record".to_vec()]);
        assert!(report.segments.is_empty());
        assert_eq!(report.unparseable_records, 1);
        assert_eq!(report.exit_code(), 2);
    }

    #[test]
    fn garbage_beside_a_green_chain_stays_exit_2_without_a_violation() {
        let op = operation("op-plus-garbage");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let mut input = blobs(&chain);
        input.push(b"this is not a record".to_vec());
        let report = validate_journal(&input);
        assert_eq!(report.segments.len(), 1);
        assert_eq!(rule(&report, 0, "C1").verdict, Verdict::Pass);
        assert_eq!(report.unparseable_records, 1);
        assert_eq!(
            report.exit_code(),
            2,
            "no violation was proven, but the evidence was partially unreadable"
        );
    }

    #[test]
    fn an_empty_journal_is_evidence_insufficient() {
        let report = validate_journal::<Vec<u8>>(&[]);
        assert_eq!(report.exit_code(), 2);
    }

    // -----------------------------------------------------------------------------------------
    // batch 3 · SessionLog input plane (C6/C8 land in S4b/S4c)
    // -----------------------------------------------------------------------------------------

    fn session_event(value: serde_json::Value) -> Vec<u8> {
        serde_json::to_vec(&value).unwrap()
    }

    #[test]
    fn session_events_classify_leniently_across_host_spellings() {
        let node_attempt = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op:step:1:effect:0",
            "request_fingerprint": "fp-1",
            "route": { "routeId": "route-a", "provider": "p" },
            "status": "success"
        }));
        let py_attempt = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op:step:2:effect:0",
            "route": { "route_id": "route-b" }
        }));
        let run_started = session_event(json!({
            "kind": "run_started",
            "run_id": "run-1",
            "route": { "routeId": "route-a" }
        }));
        let node_measured = session_event(json!({
            "kind": "prompt_measured",
            "turn": 1,
            "effect_id": "op:step:1:effect:0",
            "measurement": { "requestFingerprint": "fp-1", "inputTokens": 10 }
        }));
        let py_measured = session_event(json!({
            "kind": "prompt_measured",
            "measurement": { "request_fingerprint": "fp-2" }
        }));
        let llm_completed = session_event(json!({
            "kind": "llm_completed",
            "effect_id": "op:step:2:effect:0",
            "invocation_id": "op:step:1:effect:0"
        }));
        let unknown_kind = session_event(json!({ "kind": "compressed", "turn": 3 }));
        let kindless = session_event(json!({ "turn": 3 }));

        assert_eq!(
            classify_session_event(&node_attempt),
            Some(EvidenceEvent::ProviderAttempt {
                effect_id: Some("op:step:1:effect:0".to_string()),
                request_fingerprint: Some("fp-1".to_string()),
                route_id: Some("route-a".to_string()),
                status: Some("success".to_string()),
            })
        );
        assert_eq!(
            classify_session_event(&py_attempt),
            Some(EvidenceEvent::ProviderAttempt {
                effect_id: Some("op:step:2:effect:0".to_string()),
                request_fingerprint: None,
                route_id: Some("route-b".to_string()),
                status: None,
            })
        );
        assert_eq!(
            classify_session_event(&run_started),
            Some(EvidenceEvent::RunStarted {
                route_id: Some("route-a".to_string())
            })
        );
        assert_eq!(
            classify_session_event(&node_measured),
            Some(EvidenceEvent::PromptMeasured {
                effect_id: Some("op:step:1:effect:0".to_string()),
                request_fingerprint: Some("fp-1".to_string()),
            })
        );
        assert_eq!(
            classify_session_event(&py_measured),
            Some(EvidenceEvent::PromptMeasured {
                effect_id: None,
                request_fingerprint: Some("fp-2".to_string()),
            })
        );
        assert_eq!(
            classify_session_event(&llm_completed),
            Some(EvidenceEvent::LlmCompleted {
                effect_id: Some("op:step:2:effect:0".to_string()),
                invocation_id: Some("op:step:1:effect:0".to_string()),
            })
        );
        assert_eq!(
            classify_session_event(&unknown_kind),
            Some(EvidenceEvent::Other),
            "unknown kinds are parseable but ignored — the vocabulary evolves"
        );
        assert_eq!(classify_session_event(&kindless), Some(EvidenceEvent::Other));
        assert_eq!(
            classify_session_event(b"not json"),
            None,
            "a non-object event blob is unparseable input, never a violation"
        );
    }

    #[test]
    fn dual_input_with_a_green_journal_and_real_events_stays_green() {
        let op = operation("op-dual-green");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let stream = vec![
            session_event(json!({
                "kind": "run_started",
                "run_id": "r1",
                "route": { "routeId": "route-a" }
            })),
            session_event(json!({
                "kind": "prompt_measured",
                "turn": 1,
                "effect_id": "op-dual-green:step:1:effect:0",
                "measurement": { "requestFingerprint": "fp-1", "inputTokens": 10 }
            })),
            session_event(json!({
                "kind": "provider_attempt",
                "effect_id": "op-dual-green:step:1:effect:0",
                "request_fingerprint": "fp-1",
                "route": { "routeId": "route-a" },
                "status": "success"
            })),
            session_event(json!({
                "kind": "llm_completed",
                "turn": 1,
                "effect_id": "op-dual-green:step:1:effect:0",
                "invocation_id": "op-dual-green:step:1:effect:0"
            })),
        ];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(report.session_events, Some(4));
        assert_eq!(report.unparseable_events, 0);
        for id in ["C6.1", "C6.2", "C6.3", "C8"] {
            assert_eq!(
                cross(&report, id).verdict,
                Verdict::Pass,
                "{id}: {}",
                cross(&report, id).detail
            );
        }
        assert!(
            !report.deferred.iter().any(|line| line.starts_with("c6.") || line.starts_with("c8.")),
            "C6/C8 are implemented — the interim scope notes are gone"
        );
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn journal_only_validation_carries_no_session_plane() {
        let op = operation("op-journal-only");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let report = validate_journal(&blobs(&chain));
        assert_eq!(report.session_events, None);
        assert_eq!(report.unparseable_events, 0);
        assert_eq!(report.deferred.len(), 2, "batch-1 deferred scope is unchanged");
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn an_empty_session_plane_is_evidence_insufficient() {
        let op = operation("op-empty-session");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let report = validate_with_session_log(&blobs(&chain), &[Vec::<Vec<u8>>::new()]);
        assert_eq!(report.session_events, Some(0));
        assert!(!report.has_violations(), "an empty log proves nothing either way");
        assert_eq!(report.exit_code(), 2);
    }

    #[test]
    fn garbage_session_events_are_evidence_insufficient_not_guilty() {
        let op = operation("op-garbage-session");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let stream = vec![
            session_event(json!({ "kind": "run_started", "run_id": "r1" })),
            b"this is not an event".to_vec(),
        ];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(report.session_events, Some(1));
        assert_eq!(report.unparseable_events, 1);
        assert!(!report.has_violations());
        assert_eq!(report.exit_code(), 2);
    }

    // -----------------------------------------------------------------------------------------
    // C6 · SessionLog↔journal cross-verification
    // -----------------------------------------------------------------------------------------

    /// A green chain whose step-1 call_provider effect was resolved, plus a matching honest
    /// session stream: run pinned to route-a, the measurement, the attempt.
    fn honest_dual_input(op_name: &str) -> (Vec<KernelRecord>, Vec<Vec<u8>>) {
        let op = operation(op_name);
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_overflow_envelope(&op, 1),
        ]);
        let effect = format!("{op_name}:step:1:effect:0");
        let stream = vec![
            session_event(json!({
                "kind": "run_started",
                "run_id": "r1",
                "route": { "routeId": "route-a" }
            })),
            session_event(json!({
                "kind": "prompt_measured",
                "turn": 1,
                "effect_id": effect,
                "measurement": { "requestFingerprint": "fp-1", "inputTokens": 10 }
            })),
            session_event(json!({
                "kind": "provider_attempt",
                "effect_id": effect,
                "request_fingerprint": "fp-1",
                "route": { "routeId": "route-a" },
                "status": "success"
            })),
        ];
        (chain, stream)
    }

    #[test]
    fn an_attempt_naming_an_effect_the_replan_never_published_fails_c6() {
        let (chain, mut stream) = honest_dual_input("op-forged-effect");
        stream[2] = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op-forged-effect:step:1:effect:7",
            "request_fingerprint": "fp-1",
            "route": { "routeId": "route-a" },
            "status": "success"
        }));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C6.1").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C6.1").detail.contains("never published"),
            "{}",
            cross(&report, "C6.1").detail
        );
        assert_eq!(report.exit_code(), 1, "the forged attempt turns the run red");
    }

    #[test]
    fn an_attempt_without_an_effect_id_fails_c6_as_forged_evidence() {
        let (chain, mut stream) = honest_dual_input("op-keyless-attempt");
        stream[2] = session_event(json!({
            "kind": "provider_attempt",
            "request_fingerprint": "fp-1",
            "route": { "routeId": "route-a" },
            "status": "success"
        }));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C6.1").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C6.1").detail.contains("without effect_id"),
            "{}",
            cross(&report, "C6.1").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    #[test]
    fn an_attempt_past_the_journal_tip_degrades_c6_instead_of_failing() {
        let (chain, mut stream) = honest_dual_input("op-prefix-attempt");
        stream[2] = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op-prefix-attempt:step:9:effect:0",
            "request_fingerprint": "fp-1",
            "route": { "routeId": "route-a" },
            "status": "success"
        }));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(
            cross(&report, "C6.1").verdict,
            Verdict::Degraded,
            "a journal prefix cannot disprove an effect past its tip: {}",
            cross(&report, "C6.1").detail
        );
        assert_eq!(report.exit_code(), 0, "degradation never turns the run red");
    }

    #[test]
    fn an_attempt_on_an_operation_without_a_segment_degrades_c6() {
        let (chain, mut stream) = honest_dual_input("op-subset-journal");
        stream[2] = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op-elsewhere:step:1:effect:0",
            "request_fingerprint": "fp-1",
            "route": { "routeId": "route-a" },
            "status": "success"
        }));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C6.1").verdict, Verdict::Degraded);
        assert!(
            cross(&report, "C6.1").detail.contains("no journal segment"),
            "{}",
            cross(&report, "C6.1").detail
        );
    }

    #[test]
    fn an_orphan_fingerprint_fails_c6() {
        let (chain, mut stream) = honest_dual_input("op-orphan-fp");
        stream.remove(1); // drop the prompt_measured — the attempt's fingerprint is orphaned
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C6.2").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C6.2").detail.contains("fp-1"),
            "{}",
            cross(&report, "C6.2").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    #[test]
    fn an_attempt_without_a_fingerprint_fails_c6() {
        let (chain, mut stream) = honest_dual_input("op-fpless-attempt");
        stream[2] = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op-fpless-attempt:step:1:effect:0",
            "route": { "routeId": "route-a" },
            "status": "success"
        }));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C6.2").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C6.2")
                .detail
                .contains("without request_fingerprint"),
            "{}",
            cross(&report, "C6.2").detail
        );
    }

    #[test]
    fn an_in_run_route_change_fails_c6() {
        let (chain, mut stream) = honest_dual_input("op-route-flip");
        stream[2] = session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op-route-flip:step:1:effect:0",
            "request_fingerprint": "fp-1",
            "route": { "routeId": "route-b" },
            "status": "success"
        }));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C6.3").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C6.3").detail.contains("in-run route change"),
            "{}",
            cross(&report, "C6.3").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    #[test]
    fn a_cross_resume_route_change_degrades_c6_per_q3() {
        let (chain, mut stream) = honest_dual_input("op-route-resume");
        // A new run_started pins route-b; its attempt follows honestly. The route CHANGE
        // across the resume is degraded-marked (adapter upgrades are legal), never failed.
        stream.push(session_event(json!({
            "kind": "run_started",
            "run_id": "r1",
            "route": { "routeId": "route-b" }
        })));
        stream.push(session_event(json!({
            "kind": "prompt_measured",
            "turn": 2,
            "effect_id": "op-route-resume:step:1:effect:0",
            "measurement": { "requestFingerprint": "fp-2", "inputTokens": 11 }
        })));
        stream.push(session_event(json!({
            "kind": "provider_attempt",
            "effect_id": "op-route-resume:step:1:effect:0",
            "request_fingerprint": "fp-2",
            "route": { "routeId": "route-b" },
            "status": "success"
        })));
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(
            cross(&report, "C6.3").verdict,
            Verdict::Degraded,
            "cross-resume route changes mark, they do not fail: {}",
            cross(&report, "C6.3").detail
        );
        assert!(
            cross(&report, "C6.3").detail.contains("cross-resume"),
            "{}",
            cross(&report, "C6.3").detail
        );
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn a_pre_0_2_63_log_without_attempts_degrades_every_c6_clause() {
        let op = operation("op-old-log");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let stream = vec![
            session_event(json!({ "kind": "run_started", "run_id": "r1" })),
            session_event(json!({ "kind": "llm_completed", "turn": 1, "content": "done" })),
        ];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        for id in ["C6.1", "C6.2", "C6.3", "C8"] {
            assert_eq!(
                cross(&report, id).verdict,
                Verdict::Degraded,
                "{id}: old logs degrade (C7), never fail — {}",
                cross(&report, id).detail
            );
        }
        assert_eq!(report.exit_code(), 0);
    }

    // -----------------------------------------------------------------------------------------
    // C8 · invocation chain adjacency
    // -----------------------------------------------------------------------------------------

    /// A live chain whose first provider call overflows and whose retry completes:
    /// step 1 publishes `step:1:effect:0` (overflowed), its resolution's step publishes the
    /// retry `step:2:effect:0` (completed). The honest llm_completed for this invocation is
    /// `invocation_id = step:1:effect:0`, `effect_id = step:2:effect:0`. The operation starts
    /// with history so the compaction ladder can actually recover from the overflow.
    fn overflow_retry_chain(op_name: &str) -> Vec<KernelRecord> {
        let op = operation(op_name);
        live_chain(&[
            configure_envelope(&op),
            agent_start_with_history_envelope(&op, 14),
            resolve_overflow_envelope(&op, 1),
            resolve_completed_envelope(&op, "in-resolve-2", 1_700_000_003_000, 2, false),
        ])
    }

    #[test]
    fn an_honest_overflow_retry_invocation_passes_c8() {
        let chain = overflow_retry_chain("op-c8-green");
        let stream = vec![
            session_event(json!({ "kind": "run_started", "run_id": "r1" })),
            session_event(json!({
                "kind": "llm_completed",
                "turn": 1,
                "effect_id": "op-c8-green:step:2:effect:0",
                "invocation_id": "op-c8-green:step:1:effect:0"
            })),
        ];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(
            cross(&report, "C8").verdict,
            Verdict::Pass,
            "{}",
            cross(&report, "C8").detail
        );
        assert!(
            cross(&report, "C8").detail.contains("1 retried invocation(s)"),
            "{}",
            cross(&report, "C8").detail
        );
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn a_completed_chain_head_is_the_merge_forgery() {
        // The provider call completes with a tool call, so step 2 publishes an ExecuteTools
        // effect. Claiming invocation step:1:effect:0 → step:2:effect:0 merges the tool
        // execution into the provider invocation — the head COMPLETED, so nothing chains.
        let op = operation("op-c8-merged");
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_completed_envelope(&op, "in-resolve-1", 1_700_000_002_000, 1, true),
        ]);
        let stream = vec![session_event(json!({
            "kind": "llm_completed",
            "turn": 1,
            "effect_id": "op-c8-merged:step:2:effect:0",
            "invocation_id": "op-c8-merged:step:1:effect:0"
        }))];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C8").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C8").detail.contains("closes its invocation"),
            "{}",
            cross(&report, "C8").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    #[test]
    fn a_selected_effect_preceding_the_chain_head_fails_c8() {
        let chain = overflow_retry_chain("op-c8-backwards");
        let stream = vec![session_event(json!({
            "kind": "llm_completed",
            "turn": 1,
            "effect_id": "op-c8-backwards:step:1:effect:0",
            "invocation_id": "op-c8-backwards:step:2:effect:0"
        }))];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C8").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C8").detail.contains("does not follow the chain head"),
            "{}",
            cross(&report, "C8").detail
        );
    }

    #[test]
    fn a_selected_effect_the_replan_never_published_fails_c8() {
        let chain = overflow_retry_chain("op-c8-phantom");
        let stream = vec![session_event(json!({
            "kind": "llm_completed",
            "turn": 1,
            "effect_id": "op-c8-phantom:step:2:effect:9",
            "invocation_id": "op-c8-phantom:step:1:effect:0"
        }))];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(cross(&report, "C8").verdict, Verdict::Fail);
        assert!(
            cross(&report, "C8").detail.contains("never published"),
            "{}",
            cross(&report, "C8").detail
        );
        assert_eq!(report.exit_code(), 1);
    }

    #[test]
    fn a_first_try_invocation_has_no_adjacency_to_prove() {
        let op = operation("op-c8-first-try");
        let chain = live_chain(&[
            configure_envelope(&op),
            agent_start_envelope(&op),
            resolve_completed_envelope(&op, "in-resolve-1", 1_700_000_002_000, 1, false),
        ]);
        let stream = vec![session_event(json!({
            "kind": "llm_completed",
            "turn": 1,
            "effect_id": "op-c8-first-try:step:1:effect:0",
            "invocation_id": "op-c8-first-try:step:1:effect:0"
        }))];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(
            cross(&report, "C8").verdict,
            Verdict::Pass,
            "{}",
            cross(&report, "C8").detail
        );
        assert!(
            cross(&report, "C8").detail.contains("1 first-try"),
            "{}",
            cross(&report, "C8").detail
        );
        assert_eq!(report.exit_code(), 0);
    }

    #[test]
    fn an_invocation_past_the_journal_tip_degrades_c8() {
        // The journal is a prefix cut before the overflow resolution lands; the SessionLog
        // already tells the whole story. Plane lag degrades, never fails.
        let op = operation("op-c8-lag");
        let chain = live_chain(&[configure_envelope(&op), agent_start_envelope(&op)]);
        let stream = vec![session_event(json!({
            "kind": "llm_completed",
            "turn": 1,
            "effect_id": "op-c8-lag:step:2:effect:0",
            "invocation_id": "op-c8-lag:step:1:effect:0"
        }))];
        let report = validate_with_session_log(&blobs(&chain), &[stream]);
        assert_eq!(
            cross(&report, "C8").verdict,
            Verdict::Degraded,
            "{}",
            cross(&report, "C8").detail
        );
        assert_eq!(report.exit_code(), 0);
    }
}