openlatch-client 0.6.3

//! HTTP request handlers for all daemon routes.
//!
//! The ingest handler is CloudEvents v1.0.2 structured-mode aware:
//! 1. Validate Content-Type (accept application/cloudevents+json or
//!    application/cloudevents-batch+json; application/json during transition)
//! 2. Deserialize body as `EventEnvelope` (single) or `Vec<EventEnvelope>` (batch)
//! 3. For each envelope: policy evaluation, dedup check, privacy filter on
//!    `data`, audit log, cloud forward, verdict decision
//! 4. Return a VerdictResponse for the first/only event — batch mode aggregates
//!    verdicts by returning the most-restrictive (deny > approve > allow).
//!
//! Unknown `source` / `type` strings are valid — the tagged-enum lens carries
//! them through as `Unknown(String)` and the client never rejects.
//!
//! # The verdict path (D14, D15)
//!
//! This module is the **only** place `core::policy` and `core::cloud` are both
//! visible — they are leaves that may not import each other — which is why the
//! `ol*` extension stamping lives here rather than in either of them.
//!
//! The path never touches the network and never touches an async loop: it reads
//! the resident rule set through an atomic pointer (`AppState::policy`, written
//! only by `daemon::policy_poller`) and returns. A hot-swap mid-evaluation is
//! safe by construction — the poller publishes a whole new bundle, it never
//! mutates the resident one.

use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use axum::{
    body::Bytes,
    extract::State,
    http::{header::CONTENT_TYPE, HeaderMap, StatusCode},
    Json,
};

use crate::core::envelope::normalize;
use crate::core::policy::{PolicyMatch, ResidentBundle};
use crate::daemon::identity::{self, IdentitySignals};
use crate::daemon::AppState;
use crate::envelope::{
    new_event_id, AgentType, EventEnvelope, HookEventType, Verdict, VerdictResponse,
    VerdictResponseContext, VerdictResponseContextBody, VerdictResponseContextHeadline,
};
use crate::hook_output::SESSIONLESS_SUBJECT;
use crate::model_relay::session::Assurance;
use crate::privacy;
use crate::zone_eval::{
    AgentContext as ZoneAgentContext, Decision as ZoneDecision, Event as ZoneEvent, EventEnv,
    SessionFacts,
};

/// Accepted Content-Types for CloudEvents ingest.
const CT_CLOUDEVENTS_SINGLE: &str = "application/cloudevents+json";
const CT_CLOUDEVENTS_BATCH: &str = "application/cloudevents-batch+json";
const CT_JSON: &str = "application/json";

/// RAII guard that stamps activity counters on a live hook ingest:
/// `last_hook_at_unix_secs` to "now" on construction, and
/// `hooks_in_flight` += 1; the `Drop` impl decrements `hooks_in_flight`
/// even on early return / panic so the counter never leaks.
///
/// The auto-update worker reads both counters via `should_apply_now`
/// to decide whether the agent is currently active. The in-flight
/// counter exists so a long-running hook (>quiet window) keeps the
/// daemon "active" — entry-only timestamping would mark a 90 s hook
/// idle once 60 s elapsed.
struct HookActivityGuard<'a> {
    in_flight: &'a AtomicU32,
}

impl<'a> HookActivityGuard<'a> {
    fn new(state: &'a AppState) -> Self {
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        state.last_hook_at_unix_secs.store(now, Ordering::Relaxed);
        state.hooks_in_flight.fetch_add(1, Ordering::AcqRel);
        Self {
            in_flight: &state.hooks_in_flight,
        }
    }
}

impl Drop for HookActivityGuard<'_> {
    fn drop(&mut self) {
        self.in_flight.fetch_sub(1, Ordering::AcqRel);
    }
}

/// POST /hooks — CloudEvents v1.0.2 structured-mode ingest.
///
/// Accepts a single envelope (object body) or a batch (array body) per the
/// CloudEvents HTTP binding spec. Dispatches each envelope through the
/// per-event pipeline and returns a single VerdictResponse.
pub async fn ingest_cloudevent(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    body: Bytes,
) -> (StatusCode, HeaderMap, Json<VerdictResponse>) {
    let start = std::time::Instant::now();
    // Activity tracking for the auto-update worker. RAII so an early
    // return on Content-Type / parse failure still decrements
    // `hooks_in_flight`. Replay events bypass this — they don't reflect
    // live agent activity.
    let _activity_guard = HookActivityGuard::new(&state);

    let ct = headers
        .get(CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");

    let is_batch = ct.starts_with(CT_CLOUDEVENTS_BATCH);
    let is_single = ct.starts_with(CT_CLOUDEVENTS_SINGLE) || ct.starts_with(CT_JSON);
    if !is_batch && !is_single {
        return reject(
            StatusCode::UNSUPPORTED_MEDIA_TYPE,
            "unsupported Content-Type; expected application/cloudevents+json or application/cloudevents-batch+json",
            start,
        );
    }

    // Parse body — single envelope or array of envelopes.
    let envelopes: Vec<EventEnvelope> = if is_batch {
        match serde_json::from_slice::<Vec<EventEnvelope>>(&body) {
            Ok(v) if !v.is_empty() => v,
            Ok(_) => {
                return reject(
                    StatusCode::BAD_REQUEST,
                    "cloudevents-batch body must not be empty",
                    start,
                );
            }
            Err(e) => {
                return reject(
                    StatusCode::BAD_REQUEST,
                    &format!("invalid cloudevents batch body: {e}"),
                    start,
                );
            }
        }
    } else {
        match serde_json::from_slice::<EventEnvelope>(&body) {
            Ok(env) => vec![env],
            Err(e) => {
                return reject(
                    StatusCode::BAD_REQUEST,
                    &format!("invalid cloudevent body: {e}"),
                    start,
                );
            }
        }
    };

    // Process each envelope. In batch mode the response reflects the most
    // restrictive verdict (deny > approve > allow) so a single blocker halts
    // the whole batch from the agent's perspective.
    let lane = if headers.contains_key(crate::hook_output::CAPTURE_ONLY_HEADER) {
        Lane::CaptureOnly
    } else {
        Lane::Renders
    };
    // The pid of the process that spawned the Cline plugin's sandbox child. Transport metadata:
    // it decides nothing on the verdict path.
    let plugin_host_pid: Option<u32> = headers
        .get(crate::hook_output::PLUGIN_HOST_PID_HEADER)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.trim().parse().ok());
    let mut worst: Option<BatchOutcome> = None;
    let mut session_ref: Option<String> = None;
    let mut response_headers = HeaderMap::new();
    for envelope in envelopes {
        // A doctor probe is diagnostic traffic, not agent activity. Skipping it
        // HERE — before `process_envelope` — is what keeps it out of the audit
        // log, the event logger, the cloud rail, the event counter and dedup,
        // every one of which lives inside that function and nowhere else.
        //
        // `continue`, never an early `return`: this handler processes a BATCH,
        // and returning would drop every *other* envelope alongside the probe.
        // A batch that is nothing but probes then falls through to the
        // `worst.unwrap_or_else(…)` default below, which answers `Allow` — the
        // plain, well-formed reply the probe is looking for, and the only
        // response shape it ever sees.
        if is_doctor_probe(envelope.data.as_ref()) {
            continue;
        }
        let is_cline = envelope.source == AgentType::Cline;
        // A capture-only report of a call a plugin lane may also report waits
        // for it — see `CAPTURE_TWIN_GRACE`. Its answer is discarded, so it is
        // answered now and processed after the wait.
        if lane == Lane::CaptureOnly
            && matches!(envelope.type_, HookEventType::PreToolUse)
            && envelope
                .data
                .as_ref()
                .and_then(normalize::tool_use_id_of)
                .is_some()
        {
            let deferred_state = state.clone();
            tokio::spawn(async move {
                tokio::time::sleep(CAPTURE_TWIN_GRACE).await;
                let _ = process_envelope(
                    deferred_state,
                    envelope,
                    std::time::Instant::now(),
                    ProcessMode::Live,
                    Lane::CaptureOnly,
                )
                .await;
            });
            continue;
        }
        // Cline's file lane with no call id to fold by (VS Code's `PreToolUse`, and every
        // header-less `user_prompt_submit`): it waits the same grace, then folds into the
        // plugin's report of the same content if one claimed it.
        if lane == Lane::CaptureOnly
            && is_cline
            && waits_for_plugin_claim(&envelope, plugin_host_pid)
        {
            if envelope.type_ == HookEventType::UserPromptSubmit {
                // Observed at ARRIVAL: a prompt covers its run whichever lane is stored — only
                // the storing waits. Without this, a host with no plugin loaded loses its
                // `prompt_seen_ms` and the session-store backfill posts the prompt again.
                let _ = state
                    .cline_prompts
                    .observe(&envelope, chrono::Utc::now().timestamp_millis());
            }
            let deferred_state = state.clone();
            tokio::spawn(async move {
                tokio::time::sleep(CAPTURE_TWIN_GRACE).await;
                let (subject, identity) = claim_key(&envelope);
                if deferred_state.plugin_claims.consume(subject, &identity) {
                    tracing::debug!(
                        session_id = %subject,
                        event_type = %envelope.type_.as_str(),
                        "folded into plugin lane"
                    );
                    return;
                }
                let _ = process_envelope(
                    deferred_state,
                    envelope,
                    std::time::Instant::now(),
                    ProcessMode::Live,
                    Lane::CaptureOnly,
                )
                .await;
            });
            continue;
        }
        // The plugin's own prompt post (capture-only, but carrying the host pid): stored at
        // once, and it leaves a claim for the file lane's copy of the same prompt.
        if lane == Lane::CaptureOnly
            && is_cline
            && plugin_host_pid.is_some()
            && envelope.type_ == HookEventType::UserPromptSubmit
        {
            let (subject, identity) = claim_key(&envelope);
            state.plugin_claims.insert(subject, &identity);
        }
        let now_ms = chrono::Utc::now().timestamp_millis();
        // Any plugin event covers its run: the plugin's `beforeRun` posted the prompt itself.
        if is_cline && lane == Lane::Renders {
            if let Some(subject) = envelope.subject.as_deref() {
                state.cline_prompts.mark_covered(subject, now_ms);
            }
        }
        let prompt_backfill = state.cline_prompts.observe(&envelope, now_ms);
        let (
            ev_type,
            ev_id,
            ev_subject,
            dedup_hit,
            policy_match,
            zone_decision,
            hold_wait,
            delivery,
        ) = process_envelope(state.clone(), envelope, start, ProcessMode::Live, lane).await;
        // After the task end itself is processed, so its prompt never lands
        // ahead of it on the rail; spawned, so the hook's answer never waits.
        if let Some(job) = prompt_backfill {
            crate::daemon::cline_prompt_backfill::spawn(state.clone(), job);
        }
        // Which Cline host loaded the plugin — recorded AFTER the verdict, so the process-table
        // read never delays the answer.
        if let Some(pid) = plugin_host_pid.filter(|_| is_cline && lane == Lane::Renders) {
            crate::daemon::cline_plugin_seen::record_plugin_host(&state, pid, now_ms);
        }
        if dedup_hit {
            response_headers.insert(
                "x-openlatch-dedup",
                "true".parse().expect("static header value is valid"),
            );
        }
        if session_ref.is_none() {
            session_ref = ev_subject.filter(|s| !s.is_empty());
        }
        join_most_restrictive(
            &mut worst,
            BatchOutcome {
                verdict: verdict_for(&ev_type, zone_decision.as_ref(), policy_match.as_ref()),
                event_type: ev_type,
                event_id: ev_id,
                policy_match,
                zone_decision,
                hold_wait,
                delivery,
            },
        );
    }

    let outcome = worst.unwrap_or_else(|| BatchOutcome {
        // Empty batch would have been rejected above; this branch is unreachable
        // but keep fail-open semantics just in case.
        verdict: Verdict::Allow,
        event_type: HookEventType::Unknown(String::new()),
        event_id: new_event_id(),
        policy_match: None,
        zone_decision: None,
        hold_wait: None,
        delivery: crate::daemon::optimize::Delivery::default(),
    });

    let latency_ms = start.elapsed().as_secs_f64() * 1000.0;
    let mut response = match outcome.verdict {
        // A deny can only come from a policy match, so `reason` / `severity` /
        // `rule_id` are the deciding rule's own — the user story is "every
        // blocked action names the rule that blocked it".
        Verdict::Block => {
            let m = outcome.policy_match.as_ref();
            VerdictResponse::block(
                outcome.event_id,
                latency_ms,
                m.map(|m| m.reason.clone()).or_else(|| {
                    outcome
                        .zone_decision
                        .as_ref()
                        .map(|decision| decision.reason.clone())
                }),
                m.map(|m| m.severity.to_string()),
                m.map(|m| m.rule_id.clone()),
            )
        }
        Verdict::Ask => {
            let mut response = VerdictResponse::ask(outcome.event_id, latency_ms);
            response.reason = outcome
                .zone_decision
                .as_ref()
                .map(|d| d.reason.clone())
                .filter(|r| !r.is_empty());
            response
        }
        Verdict::Optimize => VerdictResponse::optimize(outcome.event_id, latency_ms),
        Verdict::Allow => VerdictResponse::allow(outcome.event_id, latency_ms),
    };
    if let Some((token, timeout_ms)) = outcome.hold_wait {
        response.schema_version = "1.2".to_string();
        response.hold_token = Some(token);
        response.hold_timeout_ms = Some(timeout_ms);
        response.defer = true;
    }
    if outcome.delivery.updated_input.is_some()
        || outcome.delivery.additional_context.is_some()
        || outcome.delivery.system_message.is_some()
    {
        response.schema_version = "1.2".to_string();
        response.updated_input = outcome.delivery.updated_input.unwrap_or_default();
        response.additional_context = outcome.delivery.additional_context;
        response.system_message = outcome.delivery.system_message;
    }

    // If the cloud's long-poll has dropped an alert for this session,
    // stamp it on the response's `context` field so the hook binary's
    // translator surfaces it. PreToolUse becomes `ask` (user-visible
    // permission prompt); SessionStart surfaces context-only (the
    // upstream spec has no native deny channel). The empty-check
    // short-circuits the per-event DashMap lookup — alerts are rare,
    // hook events are not.
    //
    // NOT on a policy deny. `attach_alert_context` unconditionally overwrites
    // `response.reason`, and `hook_output/claude_code.rs` prefers `context`
    // over `reason` on a deny — so a queued config alert co-occurring with a
    // block would show the developer the ALERT text and attribute the block to
    // the wrong thing. The deny is both the more urgent signal and the one the
    // developer must be able to act on. `pop_for_session` is deliberately not
    // called in that branch: the alert stays queued and surfaces on the next
    // event.
    //
    // Nor on a CAPTURE-ONLY lane (`CAPTURE_ONLY_HEADER`): its answer is thrown
    // away unread — Cline's file shims print their own `{}` — so popping the
    // alert there consumes it and shows it to no one. It stays queued for the
    // next event on a lane that renders context.
    let renders_context = lane == Lane::Renders;
    if renders_context
        && !matches!(outcome.verdict, Verdict::Block)
        && !state.pending_alerts.is_empty()
    {
        if let Some(subject) = session_ref.as_deref() {
            if let Some(alert) = state.pending_alerts.pop_for_session(subject) {
                let surface = match outcome.event_type {
                    HookEventType::PreToolUse => "pretooluse_reason",
                    HookEventType::SessionStart => "session_start_context",
                    _ => "other",
                };
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_alert_surfaced_in_session(
                        &alert.severity,
                        surface,
                    ),
                );
                attach_alert_context(&mut response, &alert);
            }
        }
    }

    (StatusCode::OK, response_headers, Json(response))
}

/// Is this envelope one of OpenLatch's own diagnostic probes?
///
/// The client's arming diagnostic proves a hook command still resolves by
/// **running it**, with a synthetic payload on stdin. That payload reaches the
/// daemon exactly like a real tool call, so without this filter every routine
/// `openlatch status`, `doctor` and `init` would fabricate agent activity into
/// the customer's local audit trail and forward it to the platform — from
/// seven `run_all_checks` call sites, on a product whose value *is* the
/// integrity of that trail.
///
/// The predicate keys on OpenLatch's own reserved `tool_name`, never on an
/// agent, so shared code gains no per-agent branch. The literal is the same one
/// the probe sends, and the binding's
/// `synthetic_probe_uses_a_non_evaluated_tool_name` calls straight into here so
/// the two cannot drift apart.
///
/// `tool_name` is not a field on `EventEnvelope` — it lives in the agent's raw
/// payload, which the envelope carries opaquely as `data`.
pub(crate) fn is_doctor_probe(data: Option<&serde_json::Value>) -> bool {
    data.and_then(|d| d.get("tool_name"))
        .and_then(serde_json::Value::as_str)
        == Some("OpenlatchDoctorProbe")
}

/// One envelope's contribution to the batch response.
struct BatchOutcome {
    verdict: Verdict,
    event_type: HookEventType,
    event_id: String,
    /// The match that produced `verdict`, so `reason` / `severity` / `rule_id`
    /// on the response come from the rule that actually decided.
    policy_match: Option<PolicyMatch>,
    zone_decision: Option<ZoneDecision>,
    hold_wait: Option<(String, u64)>,
    delivery: crate::daemon::optimize::Delivery,
}

/// Rank in the most-restrictive-wins lattice: `block > ask > optimize > allow`.
///
/// The `allow < ask < block` order is the one this fold has always had, under
/// the pre-D14 spelling. `optimize` is new and nothing in the client produces
/// one yet (the artifact evaluator that will is a later plan); it sits above
/// `allow` because the action was changed, and below `ask` because nobody is
/// asked. Its real place in the lattice is that plan's to decide.
fn verdict_rank(verdict: Verdict) -> u8 {
    match verdict {
        Verdict::Allow => 0,
        Verdict::Optimize => 1,
        Verdict::Ask => 2,
        Verdict::Block => 3,
    }
}

/// The lattice join the module docs have always claimed: fold `candidate` into
/// `worst`, keeping the most restrictive of the two.
///
/// Strictly-greater replaces, so within one rank the earliest envelope in the
/// batch wins — the same tiebreak the pre-policy fold had, and deterministic for
/// a given batch.
fn join_most_restrictive(worst: &mut Option<BatchOutcome>, candidate: BatchOutcome) {
    match worst {
        Some(current) if verdict_rank(candidate.verdict) <= verdict_rank(current.verdict) => {}
        _ => *worst = Some(candidate),
    }
}

/// The verdict for ONE envelope.
///
/// A non-shadow policy match blocks. Everything else allows. A shadow match
/// ("would have blocked") never changes the verdict — it only annotates the
/// stored event with `olverdictshadow`.
fn verdict_for(
    event_type: &HookEventType,
    zone_decision: Option<&ZoneDecision>,
    policy_match: Option<&PolicyMatch>,
) -> Verdict {
    if policy_match.is_some_and(|m| !m.shadow) {
        return Verdict::Block;
    }
    if let Some(decision) = zone_decision {
        return decision.verdict;
    }
    match event_type {
        // The one place the D14 rename is not a spelling change. `Stop` used to
        // return `approve`, which meant "user-confirmed allow". The four frozen
        // verdicts have no such member, and `ask` means allow-and-flag — so
        // renaming this arm to `Ask` would open an ASK ground on every session
        // end. `allow` is what `approve` always meant here.
        HookEventType::Stop => Verdict::Allow,
        _ => Verdict::Allow,
    }
}

/// Attach a pending alert to a verdict response.
///
/// Stamps `context.headline` and `context.body` so the hook binary's
/// translator surfaces the alert: PreToolUse becomes
/// `permissionDecision: "ask"` with headline/body as reason; SessionStart
/// gets the same payload via `additionalContext` injection. The wire
/// `verdict` field stays at whatever it already was — this path does not
/// promote an alert to a real `ask`, and the translator picks up the
/// soft-ask intent from the presence of `context`.
fn attach_alert_context(
    response: &mut VerdictResponse,
    alert: &crate::daemon::config_monitor::PendingAlert,
) {
    let headline = clamp_str(&alert.headline, 120);
    let body = clamp_str(&alert.body, 500);
    let Some(ctx) = build_context(&headline, &body) else {
        return;
    };
    response.context = Some(ctx);
    response.reason = Some(headline);
    response.schema_version = "1.1".to_string();
}

fn build_context(headline: &str, body: &str) -> Option<VerdictResponseContext> {
    let headline = VerdictResponseContextHeadline::try_from(headline.to_string()).ok()?;
    let body = VerdictResponseContextBody::try_from(body.to_string()).ok()?;
    Some(VerdictResponseContext {
        body,
        evidence: Vec::new(),
        headline,
    })
}

fn clamp_str(s: &str, max: usize) -> String {
    if s.chars().count() <= max {
        return s.to_string();
    }
    s.chars().take(max.saturating_sub(1)).chain(['…']).collect()
}

/// Source of an envelope flowing through `process_envelope`. Live hooks
/// must NEVER block the agent on cloud-forward backpressure (fail-open),
/// so the live path uses `try_send` and drops on a full channel. Replay
/// reads from a durable on-disk queue where dropping silently would lose
/// data, so the replay path uses `send().await` to apply backpressure.
/// Which hook lane posted an envelope, as far as delivery goes.
///
/// A `CaptureOnly` lane (`CAPTURE_ONLY_HEADER` — Cline's file shims) discards
/// the daemon's answer unread, so nothing it is told is ever delivered. A
/// `Renders` lane hands the answer to the agent. Recording the verdict a lane
/// could not deliver as enforced — or the one it did as not — is a false
/// record either way.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub(crate) enum Lane {
    Renders,
    CaptureOnly,
}

/// How long a capture-only report of a tool call that carries the agent's
/// call id waits before it is processed.
///
/// Where Cline loads the plugin, its file shim and its plugin both report the
/// same call, and the shim usually lands first. Whichever claims the call id
/// is the one stored, so the shim waits: the plugin — the lane that actually
/// enforces — claims it, and the shim's copy folds. Where no plugin loads, the
/// shim claims it after the wait and is stored as before. Its answer is
/// discarded either way, so the wait costs the agent nothing. Longer than the
/// plugin's own 2.5 s budget, so a plugin that answers at all has claimed.
const CAPTURE_TWIN_GRACE: Duration = Duration::from_secs(3);
// Only the claim-and-store decision waits: the shim's HTTP answer is sent at
// once (see `ingest_cloudevent`), so no host pays this on a tool call.

/// Whether a capture-only Cline report is the file lane's copy of something the plugin lane may
/// also report with no call id to fold it by: a `PreToolUse` carrying no
/// agent call id, or a `user_prompt_submit` WITHOUT the plugin's host-pid header (with it, the
/// report IS the plugin's prompt post). The caller has already checked lane and source.
fn waits_for_plugin_claim(envelope: &EventEnvelope, plugin_host_pid: Option<u32>) -> bool {
    match envelope.type_ {
        HookEventType::PreToolUse => envelope
            .data
            .as_ref()
            .and_then(normalize::tool_use_id_of)
            .is_none(),
        HookEventType::UserPromptSubmit => plugin_host_pid.is_none(),
        _ => false,
    }
}

/// The key a plugin-lane claim is left and consumed under: the session, and the event's
/// [`dedup_identity`].
fn claim_key(envelope: &EventEnvelope) -> (&str, serde_json::Value) {
    let subject = envelope
        .subject
        .as_deref()
        .unwrap_or(crate::hook_output::SESSIONLESS_SUBJECT);
    let identity = dedup_identity(envelope.data.as_ref().unwrap_or(&serde_json::Value::Null));
    (subject, identity)
}

#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum ProcessMode {
    Live,
    Replay,
}

/// Replay entrypoint for `fallback.jsonl` entries. Delegates to
/// `process_envelope` so the privacy filter runs on pre-filter hook data
/// before it reaches the cloud — mandatory because hook-binary writes
/// bypass the filter.
pub(crate) async fn process_envelope_for_replay(
    state: Arc<AppState>,
    envelope: EventEnvelope,
    start: std::time::Instant,
) -> (HookEventType, String, Option<String>, bool) {
    // The policy match is discarded: a replayed event's verdict was returned to
    // the agent hours ago, and `process_envelope` does not even evaluate in
    // Replay mode. Re-deciding it now would be meaningless at best and a
    // fabricated verdict at worst.
    let (ev_type, ev_id, subject, dedup_hit, _policy_match, _zone_decision, _hold_wait, _delivery) =
        process_envelope(
            state,
            envelope,
            start,
            ProcessMode::Replay,
            Lane::CaptureOnly,
        )
        .await;
    (ev_type, ev_id, subject, dedup_hit)
}

/// Yield-ratio inverse: replay stalls while the channel is more than
/// `1 - 1/N` full. With N=4 that's 75% — leaving a quarter of the channel
/// always reserved for live `try_send` callers.
const REPLAY_YIELD_FILL_RATIO_INV: usize = 4;
/// Delay between replay-pacing checks. Short enough that drain is
/// responsive; long enough that we don't burn CPU on a stalled channel.
const REPLAY_BACKPRESSURE_TICK: Duration = Duration::from_millis(50);

/// Forward a replay-mode event onto the cloud channel without competing
/// with live hooks for capacity. The loop yields whenever the channel is
/// more than 75% full so live `try_send` callers always find headroom.
/// On a closed channel (worker exited / daemon shutting down) the call
/// returns without further retries — the caller's offset stays anchored.
async fn forward_replay_event(
    tx: &tokio::sync::mpsc::Sender<crate::cloud::CloudEvent>,
    mut cloud_event: crate::cloud::CloudEvent,
) {
    use tokio::sync::mpsc::error::TrySendError;
    loop {
        let max_cap = tx.max_capacity();
        let free = tx.capacity();
        if free.saturating_mul(REPLAY_YIELD_FILL_RATIO_INV) < max_cap {
            if tx.is_closed() {
                tracing::warn!("cloud channel closed during replay — worker has exited");
                return;
            }
            tokio::time::sleep(REPLAY_BACKPRESSURE_TICK).await;
            continue;
        }
        match tx.try_send(cloud_event) {
            Ok(()) => return,
            Err(TrySendError::Full(ev)) => {
                cloud_event = ev;
                tokio::time::sleep(REPLAY_BACKPRESSURE_TICK).await;
            }
            Err(TrySendError::Closed(_)) => {
                tracing::warn!("cloud channel closed during replay — worker has exited");
                return;
            }
        }
    }
}

/// Per-envelope pipeline: policy evaluation, dedup, privacy-filter, log,
/// cloud-forward.
///
/// Returns `(type, id, subject, dedup_hit, policy_match)` — the caller uses
/// subject to look up pending alerts for the session, `dedup_hit` to attach the
/// `x-openlatch-dedup` response header, and `policy_match` to fold the batch
/// verdict (the verdict itself is built by the caller, after the fold).
async fn process_envelope(
    state: Arc<AppState>,
    mut envelope: EventEnvelope,
    start: std::time::Instant,
    mode: ProcessMode,
    lane: Lane,
) -> (
    HookEventType,
    String,
    Option<String>,
    bool,
    Option<PolicyMatch>,
    Option<ZoneDecision>,
    Option<(String, u64)>,
    crate::daemon::optimize::Delivery,
) {
    let event_type = envelope.type_.clone();
    let event_id = envelope.id.clone();

    // ---- Policy evaluation (D14) -----------------------------------------
    //
    // FIRST, and deliberately so. Two orderings here are load-bearing and each
    // compiles either way:
    //
    // (a) BEFORE `privacy::filter_event_with`, which rewrites `envelope.data`
    //     in place further down. Evaluating the filtered command would glob-
    //     match redacted text, silently changing which rules fire for any
    //     command containing something the credential regex recognises — a hole
    //     no test finds unless it plants a credential-shaped string.
    //
    // (b) BEFORE the content-dedup early return. Returning no match on a dedup
    //     hit would ALLOW an identical denied command repeated inside the
    //     100 ms window — an enforcement bypass an agent reaches just by
    //     retrying. Evaluation is a pure in-memory glob scan over a small rule
    //     set, so running it first costs microseconds and closes the hole. The
    //     dedup branch still suppresses the event UPLOAD; it must never
    //     suppress the VERDICT.
    //
    // `load_full()` (not `load()`) because a `Guard` is not held across the
    // awaits below. One atomic load plus an `Arc` clone — no lock, no I/O, and
    // never a call to the platform. `handlers.rs` must not read `bundle.json`:
    // the poller owns that file and publishes through this handle.
    //
    // Replay mode does not evaluate at all: a replayed envelope's verdict was
    // returned to the agent long ago, so deciding it now could only fabricate
    // one.
    let policy_bundle: Option<Arc<Option<ResidentBundle>>> = match mode {
        ProcessMode::Live => state.policy.as_ref().map(|p| p.handle.load_full()),
        ProcessMode::Replay => None,
    };
    let resident = policy_bundle.as_deref().and_then(|b| b.as_ref());

    // One tool call, one stored event and one counted action — whichever lane
    // reports it first. Cline's plugin and its `PreToolUse` file shim both fire
    // on a host that loads the plugin, and both carry the agent's own call id.
    // Claimed HERE, before evaluation, because the claim decides two things:
    // whether this report commits session state (below) and whether it is
    // stored (the dedup return further down). Both reports are still
    // evaluated, so the lane that loses still answers with a real verdict.
    //
    // Keyed on the AGENT's id only (`normalize::tool_use_id_of`), never a
    // derived one, which two separate identical calls share — with no agent
    // id every report claims and counts, exactly as before.
    let is_call_duplicate = claim_call(&state, &envelope);

    // The plugin lane's content claim for the file lane's id-less copy of this call — with or
    // without a call id, since a file-lane event without one may pair with it.
    if mode == ProcessMode::Live
        && lane == Lane::Renders
        && envelope.source == AgentType::Cline
        && envelope.type_ == HookEventType::PreToolUse
    {
        let (subject, identity) = claim_key(&envelope);
        state.plugin_claims.insert(subject, &identity);
    }

    let policy_match = resident.and_then(|bundle| {
        crate::core::policy::evaluate(bundle, event_type.as_str(), envelope.data.as_ref())
    });
    let zone_evaluation = resident
        .and_then(|bundle| bundle.zone.as_ref().map(|zone| (bundle, zone)))
        .and_then(|(bundle, zone)| {
            let event = map_zone_event(&envelope, bundle, state.config.agent_id.as_deref())?;
            let now = std::time::Instant::now();
            let now_ms = chrono::Utc::now().timestamp_millis();
            let policy = state.policy.as_ref()?;
            let outcome = policy.sessions.evaluate_committing(
                &event.session_id,
                zone.state_layout,
                now,
                !is_call_duplicate,
                |prior| {
                    let outcome = crate::daemon::optimize::evaluate_and_dispatch(
                        zone,
                        &event,
                        envelope.source.as_str(),
                        &policy.dispatch_sessions,
                        &state.config.log_dir,
                        prior,
                        now,
                        now_ms,
                    );
                    let state = outcome.state_to_commit.clone();
                    (outcome, state)
                },
            );
            let mut decision = outcome.decision;
            if degrade_unapplied_rewrite(&mut decision) {
                // The current evaluator proposal has no per-lever mutation payload
                // beyond `steer_instruction`; never claim it was applied. Ask carries
                // the proposal until an I-5 actuator can produce a final input.
                decision.verdict = Verdict::Ask;
            }
            Some((decision, event, outcome.delivery))
        });
    let mut zone_decision = zone_evaluation
        .as_ref()
        .map(|(decision, _, _)| decision.clone());
    let delivery = zone_evaluation
        .as_ref()
        .map(|(_, _, delivery)| delivery.clone())
        .unwrap_or_default();
    let mut hold_wait = None;
    if let (Some(policy), Some(bundle), Some((_, event, _)), Some(decision), Some(request)) = (
        state.policy.as_ref(),
        resident,
        zone_evaluation.as_ref(),
        zone_decision.as_ref(),
        zone_decision
            .as_ref()
            .and_then(|decision| decision.hold.clone()),
    ) {
        if !event.tool_use_id.is_empty() {
            let host_timeout = bundle
                .hold_config
                .as_ref()
                .and_then(|hold| hold.host_timeout_s)
                .unwrap_or(900)
                .max(31) as u64;
            let max_pending = bundle
                .hold_config
                .as_ref()
                .and_then(|hold| hold.max_pending_holds)
                .unwrap_or(16)
                .max(1) as usize;
            let correlation_id = envelope
                .data
                .as_ref()
                .and_then(|data| data.get("correlation_id"))
                .and_then(serde_json::Value::as_str)
                .unwrap_or(&event_id)
                .to_string();
            let artifact = decision.artifact_id.as_deref().and_then(|artifact_id| {
                bundle.zone.as_ref().and_then(|zone| {
                    zone.artifacts
                        .iter()
                        .find(|artifact| artifact.artifact_id() == Some(artifact_id))
                })
            });
            let registration = crate::daemon::hold_queue::HoldRegistration {
                tool_use_id: event.tool_use_id.clone(),
                atom_id: decision.atom_id.clone(),
                policy_id: artifact.and_then(|artifact| artifact.envelope.policy_id.clone()),
                policy_public_id: decision.policy_public_id.clone(),
                effects: decision
                    .effects
                    .iter()
                    .filter_map(|effect| serde_json::to_value(effect).ok())
                    .collect(),
            };
            match policy.holds.register(
                registration,
                request,
                event.clone(),
                correlation_id,
                envelope.time.to_rfc3339(),
                envelope.subject.clone(),
                Duration::from_secs(host_timeout),
                max_pending,
            ) {
                crate::daemon::hold_queue::RegisterResult::Pending { timeout } => {
                    hold_wait = Some((
                        event.tool_use_id.clone(),
                        timeout.as_millis().min(u64::MAX as u128) as u64,
                    ));
                }
                crate::daemon::hold_queue::RegisterResult::Full(resolution) => {
                    if let Some(decision) = zone_decision.as_mut() {
                        decision.verdict = resolution.verdict;
                        decision.hold = None;
                    }
                }
            }
        }
    }
    if let (Some(m), Some(bundle)) = (policy_match.as_ref(), resident) {
        // Acceptance B4 greps the daemon log for the rule id, and this is the
        // only line that writes it there. Emitted on every deny AND every
        // observe shadow — the observe-mode evidence loop is the whole
        // justification for a staged rollout.
        tracing::info!(
            target: "policy",
            rule_id = %m.rule_id,
            mode = %m.mode,
            verdict = %verdict_for(&event_type, zone_decision.as_ref(), Some(m)),
            shadow = m.shadow,
            revision = bundle.revision,
            "policy decision"
        );
    }

    // The session this event's own payload named, if it named one.
    //
    // `SESSIONLESS_SUBJECT` is what the hook binary writes when the payload
    // carries no session id at all, so it is a placeholder rather than an id and
    // must not reach the registry: registering it would publish a session
    // literally called "unknown" for the model relay to attribute requests to.
    let named_session = envelope
        .subject
        .as_deref()
        .map(str::trim)
        .filter(|subject| !subject.is_empty() && *subject != SESSIONLESS_SUBJECT);

    // Model-relay (D-29): stamp the shared session registry so the model relay
    // listener resolves attribution + assurance in-process (B-2). Fires on every
    // hook that carries a session subject (SessionStart + each tool call). The
    // install_id + agent_id reuse the PII-free `agent_id` (the same identifier
    // the hook stamps); `source` is the agent platform from the envelope.
    if let (Some(install_id), Some(session_id)) = (state.config.agent_id.as_deref(), named_session)
    {
        // Signals for the attribution cascade. Read from the RAW payload — this
        // is above `privacy::filter_event_with`, for the same reason `event_cwd`
        // is (a redacted value is not the value the request will echo back).
        // Nothing raw is retained: the prompt is hashed on write, and
        // `tool_use_id` is an opaque provider id carrying no PII.
        //
        // **Two agents, one pair of key names, and that is not a coincidence
        // worth re-deriving.** These read like Claude Code's payload because
        // codex-cli names them identically: its `pre-tool-use.command.input` and
        // `post-tool-use.command.input` carry `tool_use_id`, and
        // `user-prompt-submit.command.input` carries `prompt` (schemas generated
        // into the binary; read off codex-cli 0.150.1). Codex's `tool_use_id` is
        // the model's own `call_id` passed straight through
        // (`run_pre_tool_use_hooks(…, invocation.call_id, …)`), which is exactly
        // what a Responses request carries back — so this one lookup feeds the
        // join for both agents and neither side needs an agent branch.
        let data = envelope.data.as_ref();
        let str_field = |k: &str| {
            data.and_then(|d| d.get(k))
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty())
        };
        let signals = crate::model_relay::session::SessionSignals {
            tool_use_id: str_field("tool_use_id"),
            prompt: str_field("prompt"),
            other_id: data.and_then(root_session_id_of),
        };
        state.registry.upsert_signals(
            install_id,
            install_id,
            envelope.source.as_str(),
            session_id,
            &signals,
        );
    }

    // A hook event whose payload named no session gets one here, labelled —
    // see [`session_stamp`] for the rules. It runs at the daemon rather than at
    // the egress because the egress records its assurance INSIDE `data`, which
    // is safe only for envelopes we author ourselves; a hook envelope's `data`
    // is the agent's raw payload and stays untouched.
    let (resolved_subject, session_assurance) = session_stamp(
        mode,
        named_session.is_some(),
        &state.registry,
        state.config.agent_id.as_deref().unwrap_or_default(),
        envelope.source.as_str(),
    );
    if let Some(subject) = resolved_subject {
        envelope.subject = Some(subject);
    }

    // The provider the agent's hook says is in use (Cline's legacy build stamps
    // `model.provider` on every hook; its next build sends "unknown"). Read by
    // the provider-slot wiring: an editor that loaded a relay URL and uses the
    // provider, while no request reaches that provider's endpoint, is being
    // overridden. Only the id and the time are kept.
    if let Some(provider) = envelope
        .data
        .as_ref()
        .and_then(|d| d.get("model"))
        .and_then(|m| m.get("provider"))
        .and_then(|p| p.as_str())
        .map(str::trim)
        .filter(|p| !p.is_empty() && *p != "unknown")
    {
        state.registry.note_provider(
            envelope.source.as_str(),
            provider,
            crate::hooks::provider_endpoints::now_unix(),
        );
    }

    // The working directory this event reported, read once and used twice — here
    // for identity resolution and below for project registration.
    // By shape: Cline names no `cwd` and sends `workspaceRoots`.
    let event_cwd = envelope.data.as_ref().and_then(normalize::cwd_of);

    // Identity signals (I-1, D-14). Typed schema fields, so they are set on the
    // struct here rather than injected as raw keys further down like the `ol*`
    // family — `osuser` / `gitemail` / `provideracct` are declared in
    // `schemas/event-envelope.schema.json` and the platform reads them off the
    // published contract, not off `model_extra`.
    //
    // Stamped in this function and nowhere else, for the same reason the `ol*`
    // extensions are: only hook-derived envelopes pass through here. The
    // configuration monitor, the tamper reconciler and the model relay build their
    // own CloudEvents and have no human session behind them — an `osuser` on an
    // `ai.openlatch.config.*` event would claim an attribution nobody made.
    //
    // Two things about WHERE this sits are load-bearing:
    //
    // (a) It must run BEFORE `privacy::filter_event_with` further down, which
    //     rewrites `envelope.data` in place. A redacted `cwd` is not a path, and
    //     handing one to the git lookup would silently resolve the wrong
    //     repository — or none — for any directory whose name trips the
    //     credential regex.
    //
    // (b) It sits ABOVE the content-dedup early return, so a deduplicated event
    //     still gets its session observed. That is deliberate and cheap:
    //     resolution is memoised per session and spawns at most once, so the
    //     only cost is a mutex lock, and a session whose very first event was a
    //     duplicate would otherwise not start resolving until its second.
    let capture = resident.is_some_and(|b| b.capture_identity_signals);
    let (osuser, identity) = identity_stamp(
        mode,
        capture,
        envelope.subject.as_deref(),
        event_cwd,
        envelope.source.as_str(),
        identity::os_user,
        identity::observe_session,
    );
    envelope.osuser = osuser;
    envelope.gitemail = identity.git_email;
    envelope.provideracct = identity.provider_account;

    // SessionStart events get walked against the in-memory ContentHashCache
    // and the `openlatch.declared_*` fields injected into `data` before
    // forwarding. Best-effort: failures log `OL-2007` and the un-enriched
    // envelope continues through the standard pipeline.
    //
    // Cline's `task_start` / `task_resume` are its session starts: they keep
    // their own names on the wire and gain the same `declared_*` fields, so the
    // platform sees what a Cline session was configured with, not only a
    // Claude Code one.
    let config_monitor_tx = state.config_monitor.request_tx();
    if starts_a_session(&event_type) {
        if let Err(e) = crate::daemon::config_monitor::enrich_session_start(
            &mut envelope.data,
            &state.content_hash_cache,
            config_monitor_tx.as_ref(),
        )
        .await
        {
            tracing::warn!(
                code = crate::error::ERR_INVENTORY_ENRICH_FAILED,
                error = ?e,
                "session_start enrichment failed; forwarding un-enriched"
            );
        }
    } else if let Some(data) = envelope.data.as_ref().filter(|_| event_cwd.is_some()) {
        // Non-SessionStart hook with cwd: close the daemon-started-mid-session
        // gap by registering the project on first encounter. The monitor's
        // `registered_projects` set already dedupes so repeats are no-ops.
        // A root the agent declared (Cline's `workspaceRoots`) is taken as the
        // project; a bare cwd takes the marker walk.
        crate::daemon::config_monitor::try_register_project_for(data, config_monitor_tx.as_ref())
            .await;
    }

    // Dedup by (subject, type, data-hash) — identical-content events fired
    // within 100ms land on the same dedup key. CloudEvent `id` is a retry-
    // level identifier; content dedup catches same-action duplicates.
    let raw_subject = envelope.subject.clone();
    let subject = raw_subject.clone().unwrap_or_else(|| "unknown".into());
    let type_str = envelope.type_.as_str().to_string();
    let data_for_dedup = dedup_identity(&envelope.data.clone().unwrap_or_default());
    let is_content_duplicate = state
        .dedup
        .check_and_insert(&subject, &type_str, &data_for_dedup);
    // The call-id half was claimed before evaluation — see `claim_call`.
    let is_duplicate = is_content_duplicate || is_call_duplicate;

    if is_duplicate {
        tracing::debug!(
            code = crate::error::ERR_EVENT_DEDUPED,
            session_id = %subject,
            event_type = %type_str,
            "Event deduplicated within TTL window"
        );
        // The real match rides out with the dedup hit — see (b) above.
        return (
            event_type,
            event_id,
            raw_subject,
            true,
            policy_match,
            zone_decision,
            hold_wait,
            delivery,
        );
    }

    // Unknown-variant telemetry: emit one anonymised counter per unknown
    // source or type string so we can promote high-volume unknowns to named
    // variants in future releases. No PII: only the string label + client
    // version. Fail-silent if telemetry is disabled or not initialised.
    if matches!(envelope.source, AgentType::Unknown(_)) {
        crate::telemetry::capture_hook_source_unknown(envelope.source.as_str());
    }
    if matches!(envelope.type_, HookEventType::Unknown(_)) {
        crate::telemetry::capture_hook_type_unknown(envelope.type_.as_str());
    }

    // SECURITY: Privacy filter runs BEFORE logging — never log unredacted
    // credentials. The filter operates on the opaque `data` payload.
    if let Some(data) = envelope.data.as_mut() {
        privacy::filter_event_with(data, &state.privacy_filter);
    }

    // Stamp missing-but-known metadata if the hook binary didn't set it.
    // Client version always wins over whatever the hook passed — it identifies
    // the forwarder, not the emitter.
    if envelope.os.is_none() {
        envelope.os = Some(crate::envelope::os_string().to_string());
    }
    if envelope.arch.is_none() {
        envelope.arch = Some(crate::envelope::arch_string().to_string());
    }
    // The two host attributes (I-4), stamped here with the other machine facts
    // — after the content-dedup early return and after the privacy filter,
    // neither of which reads them. `capture` was resolved once above, beside
    // the identity stamp that shares it.
    let (hostid, hostname) = host_stamp(
        mode,
        capture,
        envelope.hostid.take(),
        identity::host_id,
        identity::hostname,
    );
    envelope.hostid = hostid;
    envelope.hostname = hostname;
    // OPENLATCH_VERSION, not CARGO_PKG_VERSION: this is the wire attribute the
    // platform's fleet-readiness check reads, and the manifest version cannot
    // distinguish a `main` build from the release it post-dates (build.rs).
    envelope.clientversion = Some(env!("OPENLATCH_VERSION").to_string());
    // datacontenttype defaults to application/json for every OpenLatch event.
    if envelope.datacontenttype.is_none() {
        envelope.datacontenttype = Some("application/json".to_string());
    }
    if envelope.localipv4.is_none() {
        envelope.localipv4 = state.local_ipv4;
    }
    if envelope.localipv6.is_none() {
        envelope.localipv6 = state.local_ipv6;
    }
    if envelope.publicipv4.is_none() {
        envelope.publicipv4 = state.public_ipv4;
    }
    if envelope.publicipv6.is_none() {
        envelope.publicipv6 = state.public_ipv6;
    }

    // Log the fully-enriched envelope (non-blocking — try_send drops event
    // if channel is full). The JSONL audit line is a valid CloudEvent plus
    // the olverdict/ollatencyms extension attrs, stamped in below.
    let mut envelope_value = serde_json::to_value(&envelope).unwrap_or(serde_json::Value::Null);
    // Per-envelope truth: in a batch each stored event carries ITS OWN verdict,
    // not the batch-folded one. The fold is only what the agent is told; the
    // per-envelope value is what makes the observe-mode evidence query mean
    // anything.
    let verdict_for_stored =
        verdict_for(&event_type, zone_decision.as_ref(), policy_match.as_ref());
    let latency_snapshot_ms = start.elapsed().as_millis() as u64;
    if let Some(obj) = envelope_value.as_object_mut() {
        obj.insert(
            "olverdict".to_string(),
            serde_json::json!(verdict_for_stored.to_string()),
        );
        obj.insert(
            "ollatencyms".to_string(),
            serde_json::json!(latency_snapshot_ms),
        );

        // How the session on this envelope was arrived at: `attested` when the
        // agent's own payload named it, `inferred` when the registry had to
        // pick between live sessions, `unknown` when nothing could be
        // correlated and the subject is the install bucket. The platform reads
        // the same three values from the model relay's own rows, so one key
        // answers "is this session a fact or a correlation?" on both planes.
        if let Some(assurance) = session_assurance {
            obj.insert(
                "olsessionassurance".to_string(),
                serde_json::json!(assurance.as_str()),
            );
        }

        // THE COMMON EVENT SHAPE — see `map_data_to_common`.
        map_data_to_common(obj, &event_type);
        // Egress shape (D-29, I-4). Stamped here, onto the serialized envelope,
        // for the same reason the `ol*` family is and NOT as declared properties
        // of `event-envelope.schema.json`: a declared attribute is validated by
        // TYPE at ingest, where one bad value rejects the WHOLE event, while an
        // undeclared one lands in the consumer's extension remainder and faces a
        // per-attribute guard that can drop only the attribute at fault.
        // `schemas/PROXY-EXTENSIONS.md` is their contract now — do not re-declare
        // them in the schema to "document" them.
        //
        // Scope is the same as the identity signals': hook-derived envelopes, live
        // only. A replayed envelope describes a hook call that finished hours ago,
        // and today's route is no more its route than today's bundle revision is
        // its revision. The config monitor and the tamper reconciler build their
        // own CloudEvents that never pass through here.
        if mode == ProcessMode::Live {
            stamp_egress_extensions(obj, egress_stamp(&state.egress.snapshot()));
        }
        // The six `ol*` policy extensions (D15). Only when the client is
        // actually participating in policy — with `[policy] enabled = false`
        // there is no policy state to report and stamping `olpolicyoffline`
        // would claim one. Replay never reaches here with a bundle (see the
        // evaluation block above), so a replayed envelope carries no
        // `olpolicy*` attribute — stamping an hours-old event with today's
        // revision, age and offline flag would be fabricated verdict metadata.
        if let Some(policy) = state.policy.as_ref() {
            if mode == ProcessMode::Live {
                stamp_policy_extensions(
                    obj,
                    policy_match.as_ref(),
                    resident,
                    !policy.last_fetch_ok.load(Ordering::Relaxed),
                    SystemTime::now(),
                );
                stamp_record_v2_if_winner(
                    obj,
                    zone_decision.as_ref(),
                    policy_match.as_ref(),
                    resident,
                    envelope.source.as_str(),
                    lane,
                    &event_type,
                    hold_wait.is_some(),
                    Some(&delivery),
                );
            }
        }
    }
    let envelope_json = serde_json::to_string(&envelope_value).unwrap_or_default();
    state.event_logger.log(envelope_json);

    state.event_counter.fetch_add(1, Ordering::Relaxed);

    // CLOUD-01/CLOUD-09: forward to the cloud worker. Live hooks use
    // non-blocking `try_send` so a saturated channel never stalls the
    // agent (fail-open). Replay events come from the durable fallback
    // queue and must NOT be dropped silently — replay yields to live
    // events when the channel is more than 75% full, then `try_send`
    // into whatever capacity remains.
    //
    // D15 — a LIVE envelope goes to the cloud with the `ol*` attributes
    // already stamped above, which inverts the intent of the comment that
    // used to sit here ("the platform stamps those itself"). That was true
    // while the platform was the verdict producer. The client is now the only
    // verdict producer — the platform's ingest returns a hardcoded fail-open
    // `allow` — so a verdict that is not stamped here does not exist anywhere,
    // and the observe-mode evidence loop that justifies staged rollout has
    // nothing to query.
    //
    // A REPLAYED envelope is forwarded unstamped, exactly as before: its
    // verdict and latency belong to a hook call that finished hours ago.
    //
    // Scope is deliberately narrow: HOOK-DERIVED envelopes only. The
    // configuration monitor and the tamper reconciler push their own
    // CloudEvents through `cloud_tx` and are NOT stamped — no hook action, no
    // policy evaluated, and an `olverdict` on an `ai.openlatch.config.*` event
    // would be a verdict nobody produced. Do not "complete" the coverage.
    if let Some(tx) = &state.cloud_tx {
        let agent_id = state.config.agent_id.as_deref().unwrap_or_default();
        if agent_id.is_empty() {
            tracing::warn!(
                code = "OL-1200",
                "cloud forward skipped: [daemon].agent_id is empty in config.toml — run 'openlatch init' to populate it"
            );
        } else {
            let cloud_event = crate::cloud::CloudEvent {
                envelope: match mode {
                    ProcessMode::Live => envelope_value,
                    // A replayed envelope must not carry the client's CURRENT
                    // state: its verdict was returned to the agent hours ago,
                    // and an `olpolicy*` stamped now would describe a bundle
                    // this event never saw. So the body is rebuilt from the raw
                    // envelope instead of reusing the enriched one.
                    //
                    // The agent facts are the exception, and re-stamping them
                    // is what makes `map_data_to_common`' own contract true —
                    // they are read from `data`, which is as true hours later
                    // as it was when the hook fired. Rebuilding without them
                    // threw away work that had already been done: a fail-open
                    // event reached the platform in the agent's own shape, and
                    // for an agent that nests
                    // those fields the platform cannot recover them from the
                    // raw payload either. Every event that failed open was
                    // also, permanently, an unnormalized one.
                    ProcessMode::Replay => replay_cloud_body(&envelope, &event_type),
                },
                agent_id: agent_id.to_string(),
            };
            match mode {
                ProcessMode::Live => {
                    let max_cap = tx.max_capacity();
                    match tx.try_send(cloud_event) {
                        Ok(()) => {}
                        Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                            if let Some(cs) = state.cloud_state.as_ref() {
                                cs.record_live_drop();
                            }
                            tracing::warn!(code = "OL-1200", "cloud channel full - event dropped");
                        }
                        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
                            tracing::warn!("cloud channel closed - worker has exited");
                        }
                    }
                    // Sample channel depth AFTER the try_send so the gauge
                    // reflects the in-flight count this hook left behind.
                    // One atomic load + (at most one) store on the hot path
                    // — well within the <5 ms client budget.
                    if let Some(cs) = state.cloud_state.as_ref() {
                        if max_cap > 0 {
                            let free = tx.capacity();
                            let depth = max_cap.saturating_sub(free);
                            let pct = (depth as u64).saturating_mul(100) / max_cap as u64;
                            cs.note_channel_depth_pct(pct);
                        }
                    }
                }
                ProcessMode::Replay => {
                    forward_replay_event(tx, cloud_event).await;
                }
            }
        }
    }

    if matches!(event_type, HookEventType::SessionEnd) {
        if let (Some(policy), Some(subject)) = (state.policy.as_ref(), raw_subject.as_deref()) {
            policy.sessions.remove(subject);
            policy.dispatch_sessions.remove(subject);
        }
    }
    (
        event_type,
        event_id,
        raw_subject,
        false,
        policy_match,
        zone_decision,
        hold_wait,
        delivery,
    )
}

/// Claim this report's agent-sent tool call id for its session and event type.
/// `true` when another report of the same call already holds it.
fn claim_call(state: &AppState, envelope: &EventEnvelope) -> bool {
    let Some(call_id) = envelope.data.as_ref().and_then(normalize::tool_use_id_of) else {
        return false;
    };
    let subject = envelope.subject.as_deref().unwrap_or("unknown");
    state
        .dedup
        .check_and_insert_call(subject, envelope.type_.as_str(), call_id)
}

/// Whether this event opens (or re-opens) a session, and so is enriched with
/// the `declared_*` inventory. Claude Code's `session_start`; Cline's
/// `task_start` and `task_resume`, which keep their own names on the wire.
fn starts_a_session(event_type: &HookEventType) -> bool {
    matches!(
        event_type,
        HookEventType::SessionStart | HookEventType::TaskStart | HookEventType::TaskResume
    )
}

/// Degrade only the rewrite that actually won. A Block decision may still
/// carry another contribution's proposal; that proposal must never weaken it.
fn degrade_unapplied_rewrite(decision: &mut ZoneDecision) -> bool {
    if decision.verdict == Verdict::Optimize && decision.rewrite.is_some() {
        decision.verdict = Verdict::Ask;
        true
    } else {
        false
    }
}

/// Map an opaque hook payload into the evaluator without inspecting paths named
/// by the action. Unknown fields remain in `tool_input` and arrays are preserved.
fn map_zone_event(
    envelope: &EventEnvelope,
    bundle: &ResidentBundle,
    agent_id: Option<&str>,
) -> Option<ZoneEvent> {
    let data = envelope.data.as_ref()?;
    let object = data.as_object()?;
    // Every agent's spelling of these is read through one lens
    // (`core::envelope::normalize`), which probes the top level and then one
    // level down. The second probe is what reads Cline's file lane, where each
    // hook nests its own fields under the hook's name — `preToolUse.toolName`,
    // `postToolUse.result`.
    //
    // Read-side only: the envelope's `data` stays the agent's raw payload,
    // opaque and never rewritten (AGENTS.md, *Opaque data*). We read it
    // differently; we do not rewrite it.
    use crate::core::envelope::normalize;
    // The COMMON tool: Cline's `write_to_file` is evaluated as the `Write` it
    // is, so the effect table and the shell classifier see every agent's
    // action in one vocabulary (`normalize::common_tool_of`).
    let (tool_name, tool_input) = normalize::common_tool_of(data).unwrap_or_default();
    let tool_input = tool_input.unwrap_or(serde_json::Value::Null);
    let tool_result = normalize::common_tool_response_of(data).or_else(|| {
        matches!(envelope.type_, HookEventType::PostToolUseFailure).then(|| {
            serde_json::json!({
                "is_error": true,
                "error": object.get("error").cloned().unwrap_or(serde_json::Value::Null),
                "is_interrupt": object.get("is_interrupt").cloned().unwrap_or(serde_json::Value::Bool(false)),
            })
        })
    });
    let cwd = normalize::cwd_of(data).map(str::to_string);
    let environment = bundle
        .zone
        .as_ref()
        .and_then(|z| z.meta.as_ref())
        .and_then(|m| m.environment.as_ref())
        .map(ToString::to_string);
    Some(ZoneEvent {
        event_type: match envelope.type_ {
            HookEventType::SessionEnd => "stop".to_string(),
            _ => envelope.type_.as_str().to_string(),
        },
        session_id: envelope.subject.clone().unwrap_or_default(),
        tool_use_id: normalize::common_tool_use_id_of(data).unwrap_or_default(),
        tool_name,
        tool_input,
        tool_result,
        agent: Some(ZoneAgentContext {
            agent_id: envelope
                .agentid
                .clone()
                .or_else(|| agent_id.map(str::to_string)),
            agent_row_id: None,
            agent_type: bundle
                .zone
                .as_ref()
                .and_then(|zone| zone.meta.as_ref())
                .and_then(|meta| meta.agent_type.clone())
                .map(|category| category.to_string()),
            environment,
            function: bundle.agent_function.as_ref().map(ToString::to_string),
            principal: None,
        }),
        binding: serde_json::json!({"kind": format!("{}_hook", envelope.source.as_str().replace('-', "_")), "mode": "command"}),
        env: Some(EventEnv {
            home: dirs::home_dir().map(|path| path.to_string_lossy().into_owned()),
            cwd,
            path_dirs: std::env::var_os("PATH")
                .map(|p| {
                    std::env::split_paths(&p)
                        .map(|p| p.to_string_lossy().into_owned())
                        .collect()
                })
                .unwrap_or_default(),
            additional_dirs: Vec::new(),
        }),
        session: object
            .get("session")
            .cloned()
            .and_then(|v| serde_json::from_value::<SessionFacts>(v).ok()),
        spend_delta: crate::daemon::spend::from_payload(data),
    })
}

#[allow(clippy::too_many_arguments)]
fn stamp_record_v2(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    decision: &ZoneDecision,
    bundle: Option<&ResidentBundle>,
    source: &str,
    lane: Lane,
    event_type: &HookEventType,
    deferred: bool,
    delivery: Option<&crate::daemon::optimize::Delivery>,
) {
    let scalar = |obj: &mut serde_json::Map<String, serde_json::Value>,
                  key: &str,
                  value: Option<serde_json::Value>| {
        if let Some(value) = value {
            obj.insert(key.to_string(), value);
        }
    };
    scalar(
        obj,
        "olatomid",
        decision.atom_id.clone().map(serde_json::Value::String),
    );
    scalar(
        obj,
        "olpolicyruleid",
        decision.artifact_id.clone().map(serde_json::Value::String),
    );
    let artifact = decision.artifact_id.as_deref().and_then(|artifact_id| {
        bundle
            .and_then(|resident| resident.zone.as_ref())
            .and_then(|zone| {
                zone.artifacts
                    .iter()
                    .find(|artifact| artifact.artifact_id() == Some(artifact_id))
            })
    });
    scalar(
        obj,
        "olpolicyid",
        artifact
            .and_then(|artifact| artifact.envelope.policy_id.clone())
            .map(serde_json::Value::String),
    );
    scalar(
        obj,
        "oldimension",
        decision
            .dimension
            .as_ref()
            .map(|v| serde_json::json!(v.to_string())),
    );
    scalar(
        obj,
        "olmode",
        decision
            .mode
            .as_ref()
            .map(|v| serde_json::json!(v.to_string())),
    );
    scalar(obj, "oltier", decision.tier.map(serde_json::Value::from));
    scalar(
        obj,
        "ollayer",
        artifact
            .and_then(|artifact| artifact.envelope.zone_layer.as_ref())
            .and_then(|layer| layer.node_id.clone())
            .map(serde_json::Value::String),
    );
    scalar(
        obj,
        "olspechash",
        artifact
            .and_then(|artifact| artifact.envelope.spec_hash.clone())
            .map(serde_json::Value::String),
    );
    obj.insert(
        "olbinding".into(),
        serde_json::json!(binding_of(source, lane)),
    );
    let (enforced, result) = delivery
        .and_then(|delivery| delivery.enforced.zip(delivery.result))
        .unwrap_or_else(|| delivery_outcome(source, lane, event_type, decision, deferred));
    obj.insert("olenforced".into(), serde_json::json!(i64::from(enforced)));
    obj.insert("olresult".into(), serde_json::json!(result));
    if let Some(optimize) = &decision.optimize {
        obj.insert(
            "oloptimize".into(),
            serde_json::to_value(optimize).unwrap_or_default(),
        );
        if let Some(selected) = optimize.actual.as_ref().or(optimize.monitor.as_ref()) {
            obj.insert(
                "ollever".into(),
                serde_json::json!(selected.lever.to_string()),
            );
        }
    }
    if delivery.is_some_and(|delivery| delivery.second_optimize_degraded) {
        obj.insert(
            "oloptimizedegraded".into(),
            serde_json::json!("second_optimize"),
        );
    }
    obj.insert(
        "oleffects".into(),
        serde_json::Value::String(
            serde_json::to_string(&decision.effects).unwrap_or_else(|_| "[]".into()),
        ),
    );
    obj.insert(
        "olinconclusive".into(),
        serde_json::Value::String(
            serde_json::to_string(&decision.inconclusive_facts).unwrap_or_else(|_| "[]".into()),
        ),
    );
    obj.insert(
        "olundecided".into(),
        serde_json::json!(i64::from(decision.undecided)),
    );
    if let Some(rewrite) = &decision.rewrite {
        obj.insert(
            "ollever".into(),
            serde_json::json!(rewrite.lever.to_string()),
        );
    }
    if let Some(shadow) = decision.would_have_verdict {
        obj.insert(
            "olverdictshadow".into(),
            serde_json::json!(shadow.to_string()),
        );
    }
}

/// What the target hook could actually carry, distinct from the evaluator's
/// verdict. Post-action blocks and monitor matches are evidence, not enforcement.
/// The binding a lane decided through: `{agent}_hook` for a hook lane, and
/// `cline_plugin` for the one lane that is not a hook — Cline's enforcement
/// plugin, the only Cline lane that renders an answer.
fn binding_of(source: &str, lane: Lane) -> String {
    match (source, lane) {
        ("cline", Lane::Renders) => "cline_plugin".to_string(),
        _ => format!("{}_hook", source.replace('-', "_")),
    }
}

fn delivery_outcome(
    source: &str,
    lane: Lane,
    event_type: &HookEventType,
    decision: &ZoneDecision,
    deferred: bool,
) -> (bool, &'static str) {
    let enforcing = decision
        .mode
        .as_ref()
        .is_some_and(|mode| mode.0 == "enforce");
    if deferred {
        return (enforcing, "deferred");
    }
    if decision.would_have_verdict.is_some() && !enforcing {
        return (false, "flagged");
    }
    if !enforcing || decision.artifact_id.is_none() {
        return (false, result_for(decision.verdict));
    }
    let pre_action = matches!(event_type, HookEventType::PreToolUse);
    // What the lane's translator can express. A capture-only lane expresses
    // nothing. Cline's plugin expresses a deny (`skip`) and degrades an ask to
    // allow (`hook_output::cline`).
    let renders = lane == Lane::Renders;
    let asks = renders && matches!(source, "claude-code" | "codex-cli");
    let blocks = asks || (renders && source == "cline");
    match decision.verdict {
        Verdict::Allow => (true, "allowed"),
        Verdict::Ask => (pre_action && asks, "flagged"),
        Verdict::Block if pre_action && blocks => (true, "blocked"),
        Verdict::Block => (false, "allowed"),
        Verdict::Optimize => (false, "flagged"),
    }
}

/// GET /holds/{tool_use_id} is the hook's bounded wait against the daemon.
/// Until the platform ships its hold router the configured source resolves at
/// the deadline; the hook never reaches the network itself.
pub async fn wait_for_hold(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(tool_use_id): axum::extract::Path<String>,
) -> (StatusCode, Json<VerdictResponse>) {
    let Some(policy) = state.policy.as_ref() else {
        return (
            StatusCode::NOT_FOUND,
            Json(VerdictResponse::allow(new_event_id(), 0.0)),
        );
    };
    let Some(mut resolution) = policy.holds.wait(&tool_use_id).await else {
        return (
            StatusCode::NOT_FOUND,
            Json(VerdictResponse::allow(new_event_id(), 0.0)),
        );
    };
    if resolution.answer == crate::daemon::hold_queue::HoldAnswer::Retired {
        let resident = policy.handle.load_full();
        if let Some(zone) = resident
            .as_ref()
            .as_ref()
            .and_then(|bundle| bundle.zone.as_ref())
        {
            let now = std::time::Instant::now();
            let decision = policy.sessions.evaluate(
                &resolution.event.session_id,
                zone.state_layout,
                now,
                |prior| {
                    crate::zone_eval::evaluate(
                        zone,
                        &resolution.event,
                        prior,
                        chrono::Utc::now().timestamp_millis(),
                    )
                },
            );
            resolution.verdict = decision.verdict;
            resolution.result = result_for(decision.verdict);
        }
    }
    emit_hold_resolution(&state, &tool_use_id, &resolution);
    let mut response = response_for_verdict(
        resolution.verdict,
        new_event_id(),
        0.0,
        Some("OpenLatch hold resolved with a refusal".to_string()),
    );
    response.schema_version = "1.2".to_string();
    (StatusCode::OK, Json(response))
}

fn response_for_verdict(
    verdict: Verdict,
    event_id: String,
    latency_ms: f64,
    block_reason: Option<String>,
) -> VerdictResponse {
    match verdict {
        Verdict::Allow => VerdictResponse::allow(event_id, latency_ms),
        Verdict::Ask => VerdictResponse::ask(event_id, latency_ms),
        Verdict::Optimize => VerdictResponse::optimize(event_id, latency_ms),
        Verdict::Block => VerdictResponse::block(event_id, latency_ms, block_reason, None, None),
    }
}

fn result_for(verdict: Verdict) -> &'static str {
    match verdict {
        Verdict::Allow => "allowed",
        Verdict::Ask => "flagged",
        Verdict::Block => "blocked",
        Verdict::Optimize => "rewritten",
    }
}

/**
 * Build and enqueue the deliberately narrow hold follow-up. The cloud worker
 * owns authentication, retry and durable spooling; this loopback verdict path
 * does only a non-blocking channel send.
 */
fn emit_hold_resolution(
    state: &AppState,
    tool_use_id: &str,
    resolution: &crate::daemon::hold_queue::HoldResolution,
) {
    let envelope = hold_resolution_envelope(tool_use_id, resolution);
    state
        .event_logger
        .log(serde_json::to_string(&envelope).unwrap_or_default());

    let Some(tx) = state.cloud_tx.as_ref() else {
        return;
    };
    let agent_id = state.config.agent_id.as_deref().unwrap_or_default();
    if agent_id.is_empty() {
        tracing::warn!(
            code = "OL-1200",
            "hold resolution upload skipped: [daemon].agent_id is empty in config.toml"
        );
        return;
    }
    try_enqueue_hold_resolution(tx, state.cloud_state.as_ref(), agent_id, envelope);
}

fn try_enqueue_hold_resolution(
    tx: &tokio::sync::mpsc::Sender<crate::cloud::CloudEvent>,
    cloud_state: Option<&crate::cloud::CloudState>,
    agent_id: &str,
    envelope: serde_json::Value,
) {
    match tx.try_send(crate::cloud::CloudEvent {
        envelope,
        agent_id: agent_id.to_string(),
    }) {
        Ok(()) => {}
        Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
            if let Some(cloud_state) = cloud_state {
                cloud_state.record_live_drop();
            }
            tracing::warn!(
                code = "OL-1200",
                "cloud channel full - hold resolution dropped"
            );
        }
        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
            tracing::warn!("cloud channel closed - hold resolution not uploaded");
        }
    }
}

fn hold_resolution_envelope(
    tool_use_id: &str,
    resolution: &crate::daemon::hold_queue::HoldResolution,
) -> serde_json::Value {
    let mut envelope = serde_json::json!({
        "specversion": "1.0",
        "id": new_event_id(),
        "source": "openlatch-client",
        "type": "ai.openlatch.decision.resolved",
        "time": crate::daemon::hold_queue::resolution_time(&resolution.original_time),
        "data": {
            "tool_use_id": tool_use_id,
            "correlation_id": resolution.correlation_id,
        },
        "olverdict": resolution.verdict.to_string(),
        "olresult": resolution.result,
        "olenforced": 1,
        "olhostprompted": 0,
    });
    if let Some(subject) = resolution.subject.as_ref() {
        envelope["subject"] = serde_json::Value::String(subject.clone());
    }
    envelope
}

#[allow(clippy::too_many_arguments)]
fn stamp_record_v2_if_winner(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    zone_decision: Option<&ZoneDecision>,
    policy_match: Option<&PolicyMatch>,
    bundle: Option<&ResidentBundle>,
    source: &str,
    lane: Lane,
    event_type: &HookEventType,
    deferred: bool,
    delivery: Option<&crate::daemon::optimize::Delivery>,
) {
    let Some(decision) =
        zone_decision.filter(|_| !policy_match.is_some_and(|matched| !matched.shadow))
    else {
        return;
    };
    stamp_record_v2(
        obj, decision, bundle, source, lane, event_type, deferred, delivery,
    );
}

/// Stamp the six `ol*` policy extensions onto an outbound envelope (D15).
///
/// | Extension | Value | When |
/// |---|---|---|
/// | `olverdict` | the verdict actually returned | always (stamped by the caller) |
/// | `olverdictshadow` | `"block"` | observe match only |
/// | `olpolicyruleid` | the deciding `rule_id` | any match |
/// | `olpolicybundlerev` | `revision` | bundle resident |
/// | `olpolicybundleage` | `now - built_at`, clamped at 0 | bundle resident |
/// | `olpolicyoffline` | `true` when the last poll attempt failed | always |
///
/// `ollatencyms` is **not** one of the six; it is stamped by the caller exactly
/// as it was before policy existed.
///
/// With no bundle resident the two bundle fields are OMITTED rather than
/// zeroed (D29): "never fetched a bundle" and "resident bundle, empty rules"
/// must stay distinguishable in telemetry, and an `olpolicybundlerev: 0` would
/// conflate them.
///
/// CloudEvents extension names must match `^[a-z0-9]+$`
/// (`.claude/rules/envelope-format.md`), which is why these are lowercase and
/// unpunctuated on the wire. The dotted `ai.openlatch.policy.*` forms belong to
/// the platform's OTel export layer, not here.
fn stamp_policy_extensions(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    policy_match: Option<&PolicyMatch>,
    bundle: Option<&ResidentBundle>,
    offline: bool,
    now: SystemTime,
) {
    // Fleet compatibility evidence is live-only because this helper is called
    // only from the live policy branch. Values are catalogue-derived or
    // resident identity; no configuration, credential, or raw policy body is
    // exposed, and no disk/network read enters the verdict path.
    obj.insert(
        "olcontractranges".to_string(),
        serde_json::json!(crate::core::protocol::contracts::support_header_value()),
    );
    obj.insert("olcompatversion".to_string(), serde_json::json!(1));
    obj.insert(
        "olcompatobservedat".to_string(),
        serde_json::json!(chrono::Utc::now().to_rfc3339()),
    );
    if let Some(m) = policy_match {
        if m.shadow {
            obj.insert(
                "olverdictshadow".to_string(),
                serde_json::json!(Verdict::Block.to_string()),
            );
        }
        obj.insert(
            "olpolicyruleid".to_string(),
            serde_json::json!(m.rule_id.clone()),
        );
    }
    if let Some(b) = bundle {
        obj.insert(
            "olresidentbundleschema".to_string(),
            serde_json::json!(b.schema_version),
        );
        if let Some(digest) = b.digest.as_deref() {
            obj.insert(
                "olresidentbundledigest".to_string(),
                serde_json::json!(digest),
            );
        }
        obj.insert(
            "olpolicybundlerev".to_string(),
            serde_json::json!(b.revision),
        );
        obj.insert(
            "olpolicybundleage".to_string(),
            serde_json::json!(b.age_seconds(now)),
        );
    }
    obj.insert("olpolicyoffline".to_string(), serde_json::json!(offline));
}

/// Stamp the normalized agent facts onto one serialized envelope.
///
/// Rewrite `data` into the common event shape, keeping the original under
/// `raw_agent_payload`.
///
/// Three agents described one action three ways, and every consumer carried a
/// parser for all three — including the platform, whose flat reads missed
/// Cline's nested payload entirely and left those columns null. The mapping
/// now happens once, at the single egress, and what travels is the shape every
/// consumer already reads.
///
/// **Nothing is lost.** A payload that had to be remapped carries the original
/// verbatim under `raw_agent_payload`; one that was already the common shape is
/// not touched at all, and carries no copy of itself.
///
/// Applied to replayed envelopes too, unlike `olpolicy*` and the proxy
/// attributes: those describe the client's state at stamping time, which a
/// replayed event does not share. This describes the PAYLOAD, which is as true
/// hours later as it was when the hook fired.
///
/// `agentversion` is filled from the same pass: it has been declared in
/// `event-envelope.schema.json` and in the platform's column list all along and
/// nothing ever wrote it. A producer that set it already wins.
fn map_data_to_common(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    _event_type: &HookEventType,
) {
    let Some(data) = obj.get("data").cloned() else {
        return;
    };
    if !obj.contains_key("agentversion") {
        if let Some(version) = normalize::agent_version_of(&data) {
            let version = serde_json::json!(version);
            obj.insert("agentversion".to_string(), version);
        }
    }
    if let Some(mapped) = normalize::to_common(&data) {
        obj.insert("data".to_string(), mapped);
    }
}

/// The cloud body for a REPLAYED envelope: the raw event, mapped to the common
/// shape.
///
/// Rebuilt from the envelope rather than reusing the enriched value, because
/// the enriched one carries the client's state at stamping time — `olverdict`
/// was returned to the agent hours ago and an `olpolicy*` would describe a
/// bundle this event never saw.
///
/// The mapping is the exception and must survive: it reads `data`, and `data`
/// is as true hours later as it was when the hook fired. Rebuilding without it
/// discarded work already done — a fail-open event reached the platform in the
/// agent's own shape, and for an agent that nests its hook fields nothing
/// downstream could read it. Every event that failed open was also,
/// permanently, an unmapped one.
fn replay_cloud_body(envelope: &EventEnvelope, event_type: &HookEventType) -> serde_json::Value {
    let mut value = serde_json::to_value(envelope).unwrap_or_default();
    if let Some(obj) = value.as_object_mut() {
        map_data_to_common(obj, event_type);
    }
    value
}

/// What dedup compares: the ACTION this payload describes, not the bytes that
/// described it.
///
/// One tool call can arrive twice in two shapes. Cline fires both of its lanes
/// for a shell command — the enforcement plugin sends `{toolName, parameters}`
/// flat, the file hook sends the same two fields nested under `preToolUse` —
/// so a hash of the raw payload sees two different events, and both are stored,
/// forwarded and evaluated. One command, two verdicts.
///
/// Hashing the normalized tool and its arguments collapses them. A prompt is
/// the same case: Cline's plugin posts the prompt from `beforeRun` because the
/// Cline CLI never runs the `UserPromptSubmit` hook file, and if it ever does,
/// the two lanes describe one prompt in two shapes — so a prompt event is
/// identified by its prompt. Everything else keeps hashing its whole payload:
/// an event with neither has no action to normalize, and its raw body IS its
/// identity.
///
/// This narrows nothing that was already distinct — the session and the event
/// type are still part of the key upstream, and two calls that differ in tool
/// or arguments still differ here. It widens the match to payloads that mean
/// the same thing, which is what dedup was for.
///
/// **The window still bounds it.** Dedup's TTL is 100 ms, so the two lanes
/// collapse only when they land inside it. A wider window is a separate
/// decision, with its own risk, and is not taken here.
fn dedup_identity(data: &serde_json::Value) -> serde_json::Value {
    match normalize::common_tool_of(data) {
        Some((tool, input)) => serde_json::json!({
            "tool": tool,
            "input": input,
        }),
        None => match normalize::prompt_of(data) {
            Some(prompt) => serde_json::json!({ "prompt": prompt }),
            None => data.clone(),
        },
    }
}

/// The root session a payload names beside its own session id, by SHAPE: Cline puts it at
/// `sessionContext.rootSessionId` (its hook's own session is the task, `taskId`), and its
/// provider requests carry the root session instead. Read at the top level or under
/// `raw_agent_payload`, whichever this payload has.
fn root_session_id_of(data: &serde_json::Value) -> Option<&str> {
    [
        data,
        data.get("raw_agent_payload")
            .unwrap_or(&serde_json::Value::Null),
    ]
    .into_iter()
    .find_map(|d| d.get("sessionContext")?.get("rootSessionId")?.as_str())
    .map(str::trim)
    .filter(|s| !s.is_empty())
}

/// Decide this envelope's session and how sure we are of it.
///
/// Returns `(subject to replace, assurance to stamp)` — `None` for either means
/// leave it alone. `named` is whether the agent's own payload carried a session
/// id; see [`SESSIONLESS_SUBJECT`] for why a subject can be present and still
/// not be one.
///
/// Three rules, and the first is what makes the other two safe:
///
/// 1. **A session the agent named is never second-guessed.** It is `attested`
///    and the subject is untouched.
/// 2. **No named session, a live session of this agent's** → that session id,
///    with whatever assurance the cascade returned (`inferred` when it had to
///    pick between concurrent sessions).
/// 3. **No named session, nothing live** → the install id, `unknown`. A stable
///    per-install bucket, which is what the egress worker already falls back to,
///    and the platform can render `subject == agentid` as host-level. The
///    placeholder it replaces was the same string on every install in the fleet.
///
/// Scoped to the event's own `source`, so a Cline tool call is never correlated
/// onto a live Claude Code session — the agent is the one thing we are certain
/// of here, and it is the join key the platform's `(org, source, agent_id)`
/// lookup wears.
///
/// Replay resolves nothing: a replayed envelope describes a hook call that
/// finished hours ago, and today's live session is not its session. Same reason
/// replay carries no `olpolicy*`.
fn session_stamp(
    mode: ProcessMode,
    named: bool,
    registry: &crate::model_relay::session::SessionRegistry,
    install: &str,
    source: &str,
) -> (Option<String>, Option<Assurance>) {
    if mode != ProcessMode::Live {
        return (None, None);
    }
    if named {
        return (None, Some(Assurance::Attested));
    }
    let resolved = crate::model_relay::session::resolve_session_scoped(
        registry,
        install,
        &crate::model_relay::session::RequestSignals::default(),
        Some(source),
    );
    match resolved.session_id {
        // `inferred` is the ceiling for anything resolved here, and the clamp is
        // load-bearing. The registry answers `attested` when exactly one session
        // of this agent's is live — for the model relay that is a strong claim,
        // because the request itself came from an agent process. Here nothing
        // named this session: it is still a correlation, however few candidates
        // there were, and `attested` is the value the platform reads as "the
        // agent told us". A single live session makes the guess a good one, not
        // a fact.
        Some(session_id) => (
            Some(session_id),
            Some(match resolved.assurance {
                Assurance::Attested => Assurance::Inferred,
                other => other,
            }),
        ),
        None => (
            (!install.is_empty()).then(|| install.to_string()),
            Some(Assurance::Unknown),
        ),
    }
}

/// Decide the three identity attributes for one envelope (I-1 D-14).
///
/// Returns `(osuser, {gitemail, provideracct})`. Every gate that can suppress
/// capture is expressed here, in one place, cheapest first:
///
/// 1. **`ProcessMode::Replay` never stamps.** A replayed envelope describes a
///    hook call that finished hours ago; attaching today's identity to it would
///    be fabricated metadata, the same reason replay carries no `olpolicy*`.
/// 2. **No bundle, or a bundle that does not say `true`, stamps nothing.** The
///    caller folds "no resident bundle" and "`capture_identity_signals: false`"
///    into the same `false`, which is D-02's whole point: absence and refusal
///    must be indistinguishable on this path.
/// 3. **`osuser` does not need a session; the other two do.** The OS user is a
///    property of the host, so a sessionless hook still carries it. A git email
///    belongs to a working directory and a provider account to an agent
///    session, and there is nothing to key either on without a `subject`.
/// 4. **`provideracct` is Claude Code's alone.** Its resolver reads Claude
///    Code's own state file, so `source` rides along and
///    `identity::observe_session` omits the account for every other agent
///    rather than reporting one developer's login as another agent's operator.
///
/// Both resolvers arrive as parameters rather than being called directly, and
/// that is what makes this testable at all: the real `os_user` memoises into a
/// process-global `OnceLock`, so a test that reached it through the env seam
/// would freeze that value for every other test in the binary — and race the
/// `os_user` precedence test, which mutates exactly those variables.
fn identity_stamp(
    mode: ProcessMode,
    capture: bool,
    subject: Option<&str>,
    cwd: Option<&str>,
    source: &str,
    resolve_os_user: impl FnOnce() -> Option<String>,
    observe_session: impl FnOnce(&str, Option<&str>, &str) -> IdentitySignals,
) -> (Option<String>, IdentitySignals) {
    if mode != ProcessMode::Live || !capture {
        return (None, IdentitySignals::default());
    }
    let signals = match subject {
        Some(session_id) => observe_session(session_id, cwd, source),
        None => IdentitySignals::default(),
    };
    (resolve_os_user(), signals)
}

/// Decide the two host attributes for one envelope (I-4).
///
/// They look alike and are gated completely differently, which is the whole
/// reason this is one function rather than two inline expressions:
///
/// - **`hostid` is a machine fact.** Stamped in every mode, Replay included,
///   and never behind `capture_identity_signals`: it names no person, it is a
///   one-way hash, and licensing has to count the host whatever an organization
///   decides about identity capture. A replayed envelope describes an action
///   this machine really did take, so its host id is not fabricated metadata
///   the way an `osuser` or an `olverdict` would be.
/// - **`hostname` is an identity signal.** Live and capture only, exactly like
///   the three [`identity_stamp`] resolves.
///
/// A producer-supplied `hostid` is kept and the resolver is never called —
/// nothing upstream sets one today, and honouring it keeps this function
/// idempotent for any path that later does.
///
/// Both resolvers arrive as parameters for the same reason `identity_stamp`'s
/// do: the real ones memoise into a process-global `OnceLock`, so a test that
/// reached them through their env seams would freeze those values for every
/// other test in the binary.
fn host_stamp(
    mode: ProcessMode,
    capture: bool,
    existing_hostid: Option<String>,
    resolve_host_id: impl FnOnce() -> Option<String>,
    resolve_hostname: impl FnOnce() -> Option<String>,
) -> (Option<String>, Option<String>) {
    let hostid = existing_hostid.or_else(resolve_host_id);
    let hostname = if mode == ProcessMode::Live && capture {
        resolve_hostname()
    } else {
        None
    };
    (hostid, hostname)
}

/// The three egress-shape attributes an envelope carries.
///
/// Derived as a value rather than stamped in place, which is what makes the omission
/// rules directly testable: each is a property of this struct, not of a call site.
#[derive(Debug, Default, PartialEq)]
struct EgressStamp {
    /// Open strings on the wire, closed enums in here. The three are UNDECLARED
    /// CloudEvents extension attributes — `event-envelope.schema.json` names none
    /// of them, deliberately, because a declared attribute is validated by type at
    /// ingest and one bad value there rejects the whole event. Their contract is
    /// `schemas/PROXY-EXTENSIONS.md`. The client still only ever produces a value
    /// it knows — the matches below are exhaustive over its own enums — but a
    /// reader is never entitled to assume that of the next client version.
    proxytype: Option<String>,
    proxysource: Option<String>,
    proxyintercepted: Option<bool>,
}

/// Derive the egress-shape attributes from the daemon's current route.
///
/// **Shape, never address** (D-15, D-29). The snapshot holds a masked proxy URL and this
/// function deliberately does not read it: an enterprise's proxy hostname and port are
/// internal topology, and the platform's header chip needs only the kind, the provenance
/// and whether the hop is intercepted.
///
/// Three omission rules, all load-bearing on the reader's side:
///
/// - **Direct leaves all three absent.** Absence is the platform's "no chip" signal, so a
///   `direct` value would draw a chip for a host that has no proxy.
/// - **`proxysource` rides `proxytype`**, and only when a provenance was actually
///   recorded — it is nullable in the config and absent in `direct` mode.
/// - **`proxyintercepted` waits for a real TLS observation.** `None` means "not yet
///   known"; stamping `false` would claim an observation nobody made.
fn egress_stamp(snapshot: &crate::egress::EgressSnapshot) -> EgressStamp {
    use crate::egress::{ProxySource, ProxyType};

    let proxytype = match snapshot.proxy_type() {
        ProxyType::Http => "http",
        ProxyType::Https => "https",
        ProxyType::Socks5 => "socks5",
        ProxyType::Pac => "pac",
        ProxyType::Direct => return EgressStamp::default(),
    };
    EgressStamp {
        proxytype: Some(proxytype.to_string()),
        proxysource: snapshot.source.map(|source| {
            match source {
                ProxySource::Manual => "manual",
                ProxySource::Env => "env",
                ProxySource::Windows => "windows",
                ProxySource::Macos => "macos",
                ProxySource::Gnome => "gnome",
                ProxySource::Pac => "pac",
                ProxySource::Wpad => "wpad",
            }
            .to_string()
        }),
        proxyintercepted: snapshot.tls_intercepted,
    }
}

/// Write an [`EgressStamp`] onto the serialized envelope as extension attributes.
///
/// `Option::None` means the attribute is **not written at all**. That is the whole
/// contract: absence is the platform's "no chip" signal, and a `null` would be a
/// value — one the per-attribute guard on the reading side has to reject, on an
/// envelope that had nothing wrong with it.
fn stamp_egress_extensions(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    stamp: EgressStamp,
) {
    if let Some(proxytype) = stamp.proxytype {
        obj.insert("proxytype".to_string(), serde_json::json!(proxytype));
    }
    if let Some(proxysource) = stamp.proxysource {
        obj.insert("proxysource".to_string(), serde_json::json!(proxysource));
    }
    if let Some(intercepted) = stamp.proxyintercepted {
        obj.insert(
            "proxyintercepted".to_string(),
            serde_json::json!(intercepted),
        );
    }
}

fn reject(
    status: StatusCode,
    message: &str,
    start: std::time::Instant,
) -> (StatusCode, HeaderMap, Json<VerdictResponse>) {
    tracing::warn!(%message, http_status = %status, "cloudevents ingest rejected");
    let latency_ms = start.elapsed().as_secs_f64() * 1000.0;
    // Fail-open: return an allow verdict even on malformed bodies so the
    // agent never blocks on daemon-internal validation failures.
    let response = VerdictResponse::allow(new_event_id(), latency_ms);
    (status, HeaderMap::new(), Json(response))
}

/// GET /health — liveness probe, no authentication required (DAEM-05).
///
/// Returns 200 with daemon status, version, uptime in seconds, and the live
/// state of every supervised in-process subsystem:
///
/// ```json
/// { "status": "ok", "version": "…", "uptime_secs": 42,
///   "subsystems": { "model_relay": {"state": "running", "restarts": 0}, … } }
/// ```
///
/// `status` is `"degraded"` — not `"ok"` — whenever a `RestartPolicy::Always`
/// subsystem is not running. Before this, the body was a hardcoded
/// `{"status":"ok"}`: the model relay listener could fail its bind and leave every
/// agent on the machine getting ECONNREFUSED while `/health` reported perfect
/// health. It still answers **200** in the degraded case, because it is a
/// liveness probe and all three existing callers — `check_health` /
/// `wait_for_health` (`cli/commands/lifecycle.rs`) and `probe_self_health`
/// (`daemon/mod.rs`) — only test `status().is_success()`. Flipping the HTTP
/// status would make a single restarting subsystem look like a dead daemon and
/// send `openlatch start` down the duplicate-spawn path.
pub async fn health(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    let uptime_secs = state.started_at.elapsed().as_secs();
    let degraded = state.health.degraded_names();
    if !degraded.is_empty() {
        tracing::debug!(
            code = crate::error::ERR_SUBSYSTEM_DEGRADED,
            subsystems = %degraded.join(","),
            "/health reporting degraded"
        );
    }
    let egress_failed = state.egress.status() == crate::egress::EgressStatus::Failed;
    Json(serde_json::json!({
        "status": health_status(!degraded.is_empty(), egress_failed),
        // OPENLATCH_VERSION, not CARGO_PKG_VERSION. This field answers "which
        // build is this daemon running", and the manifest version cannot: it
        // reads 0.1.14 on the release AND on every `main` build after it, so a
        // freshly installed daemon reports the version of the release it
        // post-dates. Observed for real — a daemon running a rebuilt binary
        // reported 0.1.14 here while the wire `clientversion` (below, correct
        // since #136) reported 0.1.15-dev.84+gae6f8f1, so the endpoint an
        // operator checks to confirm a deploy was the one lying about it.
        "version": env!("OPENLATCH_VERSION"),
        // The PID of the process actually answering, so a caller can prove the
        // daemon it just spawned is the daemon on the port. `daemon.pid` cannot
        // settle that: a start that loses the bind race leaves the file naming
        // whichever process wrote it last, and `openlatch init` reported
        // "Daemon started (PID N)" for a child that had already exited while
        // the port was still held by a 40-hour-old process from a deleted
        // worktree. This field is the one source that cannot be stale.
        "pid": std::process::id(),
        "uptime_secs": uptime_secs,
        "subsystems": state.health.subsystems_json(),
    }))
}

/// The one-word cloud verdict `/metrics` publishes, in ladder order.
///
/// Extracted from the inline ladder it used to be so that `doctor`'s
/// every-word table test can enumerate the set: a new word reaching `doctor`
/// without an arm renders as "report this as a bug" to the operator.
///
/// `license_blocked` sits between `auth_error` and `no_credential`, and it
/// holds whenever a gate exists and no successful POST has cleared it —
/// **including after the cadence has elapsed**. That is deliberate: once
/// `until` passes the next POST is merely *due*, and reporting `connected`
/// before one has actually succeeded would tell an operator the licence
/// problem is over when nothing has checked.
fn cloud_status_word(cs: &crate::cloud::CloudState) -> &'static str {
    if cs.is_auth_error() {
        "auth_error"
    } else if cs.license_gate().is_some() {
        "license_blocked"
    } else if cs.is_no_credential() {
        "no_credential"
    } else if cs.is_emergency_mode() {
        "emergency"
    } else if cs.consecutive_drops() > 0 {
        "network_error"
    } else {
        "connected"
    }
}

/// The live licence refusal as JSON, or `null`.
///
/// `doctor` reads `code` for its remedy and `licensing_url` for where to send
/// the operator; `openlatch status --json` carries the whole object verbatim.
fn cloud_last_refusal_json(gate: Option<&crate::cloud::LicenseGate>) -> serde_json::Value {
    match gate {
        Some(gate) => serde_json::json!({
            "code": gate.code,
            "at": chrono::DateTime::<chrono::Utc>::from(gate.at).to_rfc3339(),
            "licensing_url": gate.licensing_url,
        }),
        None => serde_json::Value::Null,
    }
}

/// GET /metrics — event counters, no authentication required (DAEM-06).
///
/// Returns total events processed (deduped events excluded), uptime in seconds,
/// the available update version (if one was discovered at startup), and cloud
/// forwarding metrics (cloud_status, cloud_forwarded_count, cloud_last_sync_secs).
pub async fn metrics(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    let events = state.event_counter.load(Ordering::Relaxed);
    let uptime_secs = state.started_at.elapsed().as_secs();
    let update_available = state.get_available_update();

    let (
        cloud_status,
        cloud_forwarded_count,
        cloud_last_sync_secs,
        cloud_drop_count,
        cloud_api_url,
        cloud_consecutive_live_drops,
        cloud_emergency_mode,
        cloud_channel_high_water_ms,
    ) = if let Some(ref cs) = state.cloud_state {
        let forwarded = cs.forwarded_count();
        let last_sync = cs.last_sync_secs();
        let drops = cs.drop_count();
        let live_drops = cs.consecutive_live_drops();
        let emergency = cs.is_emergency_mode();
        let high_water_ms = cs.channel_high_water_window_ms();
        (
            cloud_status_word(cs),
            forwarded,
            last_sync,
            drops,
            state.config.cloud.api_url.as_str(),
            live_drops,
            emergency,
            high_water_ms,
        )
    } else {
        ("not_configured", 0u64, 0u64, 0u64, "", 0u64, false, 0u64)
    };
    let cloud_last_refusal = cloud_last_refusal_json(
        state
            .cloud_state
            .as_ref()
            .and_then(crate::cloud::CloudState::license_gate)
            .as_deref(),
    );
    let policy_license_refusal = cloud_last_refusal_json(
        state
            .cloud_state
            .as_ref()
            .and_then(crate::cloud::CloudState::policy_license_refusal)
            .as_deref(),
    );

    let (outbox_pending_bytes, outbox_pending_count) = match state.outbox.as_ref() {
        Some(o) => (o.byte_size(), o.pending_count()),
        None => (0u64, 0u64),
    };
    let fallback_pending_count = fallback_log_line_count();
    let policy = policy_metrics(state.policy.as_ref(), SystemTime::now());
    let mut compatibility =
        crate::core::protocol::contracts::read_compatibility(&crate::config::openlatch_dir()).ok();
    // The sidecar is best-effort disk state. In-memory validity survives later
    // transport outcomes, so a failed legacy-clear write cannot resurrect an
    // old peer acknowledgement and a pending response can retain the last
    // acknowledged selection without retaining an obsolete refusal.
    if let Some(policy) = state.policy.as_ref() {
        if let Some(compatibility) = compatibility.as_mut() {
            mask_stale_policy_compatibility(policy, compatibility);
        }
    }
    let state_evicted = state
        .policy
        .as_ref()
        .map(|policy| policy.sessions.evicted_count())
        .unwrap_or(0);
    let pending_holds = state
        .policy
        .as_ref()
        .map(|policy| policy.holds.pending_count())
        .unwrap_or(0);
    let egress = egress_metrics(&state.egress);

    // Channel depth — uses tokio mpsc's free `.capacity()` getter so no
    // sender wrapper is needed. `max_capacity` reflects the configured
    // channel size (default 1000); `capacity` is the free slot count, so
    // `max - free` is the in-flight depth.
    let (cloud_channel_depth, cloud_channel_size_max) = match state.cloud_tx.as_ref() {
        Some(tx) => {
            let max = tx.max_capacity();
            let free = tx.capacity();
            (max.saturating_sub(free) as u64, max as u64)
        }
        None => (0u64, 0u64),
    };

    let mut body = serde_json::json!({
        "events_processed": events,
        "uptime_secs": uptime_secs,
        "update_available": update_available,
        "cloud_status": cloud_status,
        "cloud_last_refusal": cloud_last_refusal,
        // A hash, not an identifier: no URL, no tenant, nothing invertible. That
        // is what lets it ride an UNAUTHENTICATED endpoint, where the masked
        // proxy URL deliberately may not.
        "host_id": identity::host_id(),
        "cloud_forwarded_count": cloud_forwarded_count,
        "cloud_last_sync_secs": cloud_last_sync_secs,
        "cloud_drop_count": cloud_drop_count,
        "cloud_api_url": cloud_api_url,
        "cloud_channel_depth": cloud_channel_depth,
        "cloud_channel_size_max": cloud_channel_size_max,
        "cloud_consecutive_live_drops": cloud_consecutive_live_drops,
        "cloud_emergency_mode": cloud_emergency_mode,
        "cloud_channel_high_water_ms": cloud_channel_high_water_ms,
        "cloud_channel_high_water_pct_trip": crate::cloud::worker::HIGH_WATER_TRIP_PCT,
        "outbox_pending_bytes": outbox_pending_bytes,
        "outbox_pending_count": outbox_pending_count,
        "fallback_pending_count": fallback_pending_count,
        "policy_enabled": policy.enabled,
        "policy_has_bundle": policy.has_bundle,
        "policy_schema_version": policy.schema_version,
        "policy_bundle_digest": policy.bundle_digest,
        "policy_agent_function": policy.agent_function,
        "policy_revision": policy.revision,
        "policy_bundle_age_seconds": policy.bundle_age_seconds,
        "policy_last_poll_ok_secs": policy.last_poll_ok_secs,
        "policy_floor_blocked": policy.floor_blocked,
        "policy_installed_version": policy.installed_version,
        "policy_min_client_version": policy.min_client_version,
        "policy_omitted_kinds": policy.omitted_kinds,
        "policy_skipped_items": policy.skipped_items,
        "contract_compatibility": compatibility,
        "policy_state_evicted": state_evicted,
        "policy_pending_holds": pending_holds,
        // Egress. NO URL KEY, ever: `/metrics` is unauthenticated and a
        // corporate proxy's address is internal topology. The masked URL lives
        // on the authenticated `/admin/egress/status` and nowhere else.
        "egress_up": egress.up,
        "egress_status": egress.status,
        "egress_consecutive_failures": egress.consecutive_failures,
        "proxy_in_use": egress.proxy_in_use,
        // Subsystem supervision. `subsystem_restarts_total` climbing is the
        // signal that something in-process keeps dying and being brought back;
        // `subsystem_degraded_count` > 0 means it is down right now.
        "subsystem_restarts_total": state.health.total_restarts(),
        "subsystem_degraded_count": state.health.degraded_count(),
    });
    // Same shape as `cloud_last_refusal`, but for the bundle fetch: set on an
    // organization that was never licensed, before any event is queued.
    // Inserted after the literal because `json!` is at its recursion limit.
    body["policy_license_refusal"] = policy_license_refusal;
    body["policy_empty_bundle"] = serde_json::json!(policy.empty_bundle);
    body["policy_resident_aip_count"] = policy.resident_aip_count;
    body["policy_last_platform_contact_at"] = policy.last_platform_contact_at;
    body["policy_last_download_verified_at"] = policy.last_download_verified_at;
    body["policy_readiness_state"] = policy.readiness_state;
    body["policy_last_attempted_at"] = policy.last_attempted_at;
    body["policy_last_outcome"] = policy.last_outcome;
    body["policy_readiness_reason"] = policy.readiness_reason;
    Json(body)
}

/// `/health`'s single aggregate word.
///
/// Egress `failed` degrades the daemon too. A client that cannot get a byte out
/// is not healthy in any sense an operator cares about, and every subsystem can
/// be perfectly `running` while that is true — the tasks are alive, they just
/// cannot reach anything. Only `failed` counts: `degraded` means the
/// configuration resolved with warnings, and a warning is not a liveness problem.
///
/// Extracted so the egress contribution is testable without standing up a whole
/// `AppState`.
fn health_status(subsystem_degraded: bool, egress_failed: bool) -> &'static str {
    if subsystem_degraded || egress_failed {
        "degraded"
    } else {
        "ok"
    }
}

/// The `/metrics` view of egress.
struct EgressMetrics {
    /// `false` once the aggregate reads `failed`. The one gauge an alert rule
    /// should key on.
    up: bool,
    /// The `egress.status` word, so a human reading the scrape can tell a
    /// configuration warning from an outage without a second lookup.
    status: &'static str,
    /// Consecutive failures since the last success.
    consecutive_failures: u32,
    /// Whether a proxy is in the path at all.
    proxy_in_use: bool,
}

/// Takes the state rather than the whole `AppState` so the privacy invariant
/// this encodes — no URL, ever — is directly testable.
fn egress_metrics(state: &crate::egress::EgressState) -> EgressMetrics {
    let status = state.status();
    EgressMetrics {
        up: status != crate::egress::EgressStatus::Failed,
        status: status.as_str(),
        consecutive_failures: state.consecutive_failures(),
        proxy_in_use: state.snapshot().proxy_in_use,
    }
}

/// The `/metrics` view of the policy plane (D48).
struct PolicyMetrics {
    /// `[policy] enabled` — whether the client participates at all. `false`
    /// looks identical to a broken feature during the canary otherwise.
    enabled: bool,
    /// **D29.** `true` = a last-known-good bundle is resident, so the host is
    /// still enforcing whatever it last activated. `false` = no bundle has ever
    /// been activated, so the host allows everything. Conflating the two is how
    /// a client silently disarms on first boot and nobody notices.
    has_bundle: bool,
    /// Effective schema and raw-byte digest of the exact resident bundle.
    schema_version: serde_json::Value,
    bundle_digest: serde_json::Value,
    /// Resident revision, or null when there is no bundle.
    revision: serde_json::Value,
    empty_bundle: bool,
    /// `now - built_at` clamped at 0 (policy freshness), or null. A large value
    /// can be perfectly healthy — it means the rules have not changed.
    bundle_age_seconds: serde_json::Value,
    /// Seconds since the last successful poll (connectivity), or null when no
    /// poll has ever succeeded. Read from the in-memory atomic the poller
    /// writes — never by re-reading `bundle.meta.json` on every scrape.
    last_poll_ok_secs: serde_json::Value,
    last_platform_contact_at: serde_json::Value,
    last_download_verified_at: serde_json::Value,
    readiness_state: serde_json::Value,
    last_attempted_at: serde_json::Value,
    last_outcome: serde_json::Value,
    readiness_reason: serde_json::Value,
    /// The resident bundle's `client_config.agent_context.function`, or null.
    ///
    /// The one field that says which agent the platform thinks this install is,
    /// and therefore whether a rule carrying `conditions` can fire here at all.
    /// Null covers every way the context can be missing — no bundle, a bundle
    /// with no `client_config`, no `agent_context`, no `function`, or a value
    /// this build cannot parse — because they are indistinguishable in effect:
    /// scoped rules match nothing. Without it, "the rule did not fire" and "the
    /// context never arrived" look identical from outside the process.
    agent_function: serde_json::Value,
    floor_blocked: bool,
    installed_version: serde_json::Value,
    min_client_version: serde_json::Value,
    omitted_kinds: serde_json::Value,
    skipped_items: usize,
    /// Distinct policies represented by successfully loaded schema-2
    /// evaluation artifacts. Null means the resident bundle is legacy or an
    /// artifact cannot be attributed without guessing. Exception artifacts do
    /// not represent an AIP being evaluated and are deliberately excluded.
    resident_aip_count: serde_json::Value,
}

fn mask_stale_policy_compatibility(
    policy: &crate::daemon::PolicyRuntime,
    compatibility: &mut crate::core::protocol::contracts::CompatibilityState,
) {
    let (selection_acknowledged, diagnostic_valid) = policy
        .readiness
        .read()
        .map(|readiness| {
            (
                readiness.selection_acknowledged,
                readiness.compatibility_diagnostic_valid,
            )
        })
        .unwrap_or((false, false));
    if !selection_acknowledged {
        compatibility
            .selections
            .remove(crate::core::protocol::contracts::POLICY_BUNDLE_FAMILY);
    }
    if !diagnostic_valid {
        compatibility
            .diagnostics
            .remove(crate::core::protocol::contracts::POLICY_BUNDLE_FAMILY);
    }
}

/// Takes the runtime rather than the whole `AppState` so the D29 distinction
/// this encodes is directly testable.
fn policy_metrics(policy: Option<&crate::daemon::PolicyRuntime>, now: SystemTime) -> PolicyMetrics {
    let Some(policy) = policy else {
        return PolicyMetrics {
            enabled: false,
            has_bundle: false,
            schema_version: serde_json::Value::Null,
            bundle_digest: serde_json::Value::Null,
            revision: serde_json::Value::Null,
            empty_bundle: false,
            bundle_age_seconds: serde_json::Value::Null,
            last_poll_ok_secs: serde_json::Value::Null,
            last_platform_contact_at: serde_json::Value::Null,
            last_download_verified_at: serde_json::Value::Null,
            readiness_state: serde_json::Value::Null,
            last_attempted_at: serde_json::Value::Null,
            last_outcome: serde_json::Value::Null,
            readiness_reason: serde_json::Value::Null,
            agent_function: serde_json::Value::Null,
            floor_blocked: false,
            installed_version: serde_json::Value::Null,
            min_client_version: serde_json::Value::Null,
            omitted_kinds: serde_json::json!([]),
            skipped_items: 0,
            resident_aip_count: serde_json::Value::Null,
        };
    };
    let resident = policy.handle.load_full();
    let (
        has_bundle,
        schema_version,
        bundle_digest,
        revision,
        bundle_age_seconds,
        agent_function,
        omitted_kinds,
        skipped_items,
        resident_aip_count,
        empty_bundle,
    ) = match resident.as_ref() {
        Some(b) => (
            true,
            serde_json::json!(b.schema_version),
            serde_json::json!(b.digest),
            serde_json::json!(b.revision),
            serde_json::json!(b.age_seconds(now)),
            // `AgentFunction`'s generated `Serialize` writes the wire spelling
            // and `None` writes `null` — which is exactly what "no context"
            // has to read as. One source for that string, not a second one
            // via `Display`.
            serde_json::json!(b.agent_function),
            serde_json::json!(b.omitted_kinds),
            b.zone.as_ref().map_or(0, |zone| zone.skipped.len()),
            resident_aip_count(b),
            b.command_rules.is_empty()
                && b.request_rules.is_empty()
                && b.zone
                    .as_ref()
                    .is_none_or(|zone| zone.artifacts.is_empty() && zone.exceptions.is_empty()),
        ),
        None => (
            false,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::json!([]),
            0,
            serde_json::Value::Null,
            false,
        ),
    };
    let floor = policy.floor.read().ok().and_then(|guard| guard.clone());
    // 0 is the "never" sentinel on the atomic, reported as null rather than as
    // "0 seconds ago" — the opposite of the truth.
    let last_poll_ok_at = policy.last_poll_ok_at.load(Ordering::Relaxed);
    let last_poll_ok_secs = if last_poll_ok_at > 0 {
        let now_unix = now
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs() as i64)
            .unwrap_or(0);
        serde_json::json!(now_unix.saturating_sub(last_poll_ok_at).max(0))
    } else {
        serde_json::Value::Null
    };
    let timestamp = |seconds: i64| {
        if seconds <= 0 {
            return serde_json::Value::Null;
        }
        chrono::DateTime::from_timestamp(seconds, 0)
            .map(|at| at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
            .map_or(serde_json::Value::Null, serde_json::Value::String)
    };
    let last_platform_contact_at = timestamp(last_poll_ok_at);
    let last_download_verified_at =
        timestamp(policy.last_download_verified_at.load(Ordering::Relaxed));
    let readiness = policy.readiness.read().ok().map(|state| state.clone());
    let readiness_state = readiness.as_ref().map_or(serde_json::Value::Null, |state| {
        serde_json::to_value(state.phase).unwrap_or(serde_json::Value::Null)
    });
    let last_attempted_at = readiness
        .as_ref()
        .and_then(|state| state.last_attempted_at)
        .map_or(serde_json::Value::Null, timestamp);
    let last_outcome = readiness
        .as_ref()
        .and_then(|state| state.last_outcome)
        .map_or(serde_json::Value::Null, |value| serde_json::json!(value));
    let readiness_reason = readiness
        .as_ref()
        .and_then(|state| state.reason_code)
        .map_or(serde_json::Value::Null, |value| serde_json::json!(value));
    PolicyMetrics {
        enabled: true,
        has_bundle,
        schema_version,
        bundle_digest,
        revision,
        empty_bundle,
        bundle_age_seconds,
        last_poll_ok_secs,
        last_platform_contact_at,
        last_download_verified_at,
        readiness_state,
        last_attempted_at,
        last_outcome,
        readiness_reason,
        agent_function,
        floor_blocked: floor.is_some(),
        installed_version: floor.as_ref().map_or(serde_json::Value::Null, |state| {
            serde_json::json!(state.installed_version)
        }),
        min_client_version: floor.as_ref().map_or(serde_json::Value::Null, |state| {
            serde_json::json!(state.minimum_version)
        }),
        omitted_kinds,
        skipped_items,
        resident_aip_count,
    }
}

/// Count policies represented by the schema-2 artifacts that actually loaded.
///
/// `policy_id` is the canonical grouping key. If any loaded artifact lacks a
/// nonempty canonical identifier, the exact total is unknowable and the metric
/// stays null rather than inferring policy identity from a display identifier.
fn resident_aip_count(resident: &crate::core::policy::ResidentBundle) -> serde_json::Value {
    resident
        .aip_count()
        .map_or(serde_json::Value::Null, |count| serde_json::json!(count))
}

/// Count of unread `fallback.jsonl` entries — i.e. lines past the persisted
/// replay offset. Used by `/metrics` to advertise pending hook-written
/// entries. Skipping the already-consumed prefix matters now that the file
/// is capped at 50 MB (drop-oldest via offset advance, not file rewrite):
/// otherwise every scrape would re-scan up to 50 MB of dead bytes.
fn fallback_log_line_count() -> u64 {
    use std::io::{BufRead, Seek, SeekFrom};
    let log_dir = crate::config::openlatch_dir().join("logs");
    let path = log_dir.join("fallback.jsonl");
    let offset_path = log_dir.join("fallback.jsonl.offset");
    let offset: u64 = std::fs::read_to_string(&offset_path)
        .ok()
        .and_then(|s| s.trim().parse().ok())
        .unwrap_or(0);
    let Ok(file) = std::fs::File::open(&path) else {
        return 0;
    };
    let mut reader = std::io::BufReader::new(file);
    if offset > 0 && reader.seek(SeekFrom::Start(offset)).is_err() {
        return 0;
    }
    reader
        .lines()
        .map_while(Result::ok)
        .filter(|l| !l.trim().is_empty())
        .count() as u64
}

/// POST /shutdown — graceful shutdown trigger, requires authentication (DAEM-07).
pub async fn shutdown_handler(State(state): State<Arc<AppState>>) -> StatusCode {
    let mut tx = state.shutdown_tx.lock().await;
    if let Some(sender) = tx.take() {
        let _ = sender.send(());
        StatusCode::OK
    } else {
        StatusCode::GONE
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::policy::test_support::{resident, rule};
    use crate::generated::types::{PolicyRuleMode, PolicyRuleSeverity};

    fn approved_hold_resolution() -> crate::daemon::hold_queue::HoldResolution {
        crate::daemon::hold_queue::HoldResolution {
            answer: crate::daemon::hold_queue::HoldAnswer::Approved,
            verdict: Verdict::Allow,
            result: "held_approved",
            event: crate::zone_eval::Event::default(),
            correlation_id: "corr-original".into(),
            original_time: "2999-09-11T12:34:56.789Z".into(),
            subject: Some("session-original".into()),
        }
    }

    #[test]
    fn timeout_hold_resolution_preserves_correlation_and_reports_no_host_prompt() {
        let mut resolution = approved_hold_resolution();
        resolution.answer = crate::daemon::hold_queue::HoldAnswer::Timeout;
        resolution.verdict = Verdict::Block;
        resolution.result = "held_timeout";
        let envelope = hold_resolution_envelope("tool-original", &resolution);
        let object = envelope.as_object().expect("CloudEvent object");
        let mut keys: Vec<_> = object.keys().map(String::as_str).collect();
        keys.sort_unstable();
        assert_eq!(
            keys,
            [
                "data",
                "id",
                "olenforced",
                "olhostprompted",
                "olresult",
                "olverdict",
                "source",
                "specversion",
                "subject",
                "time",
                "type",
            ]
        );
        assert_eq!(
            envelope["data"],
            serde_json::json!({
                "tool_use_id": "tool-original",
                "correlation_id": "corr-original",
            })
        );
        assert_eq!(envelope["subject"], "session-original");
        assert_eq!(envelope["time"], "2999-09-11T12:34:56.789Z");
        assert_eq!(envelope["olverdict"], "block");
        assert_eq!(envelope["olresult"], "held_timeout");
        assert_eq!(envelope["olenforced"], 1);
        assert_eq!(envelope["olhostprompted"], 0);
    }

    #[tokio::test]
    async fn hold_resolution_uses_the_existing_cloud_worker_channel() {
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        let envelope = hold_resolution_envelope("tool-original", &approved_hold_resolution());
        try_enqueue_hold_resolution(&tx, None, "agent-isolated", envelope.clone());

        let forwarded = rx.recv().await.expect("resolution queued");
        assert_eq!(forwarded.agent_id, "agent-isolated");
        assert_eq!(forwarded.envelope, envelope);
    }

    /// The diagnostic probe never reaches the audit trail, and this is the
    /// predicate that decides it.
    ///
    /// The skip sits in the batch loop **before** `process_envelope`, which is
    /// the sole owner of the event logger, the cloud forward, the event counter
    /// and dedup — so a `true` here IS "no audit line written and nothing
    /// forwarded". Asserted at the predicate rather than through a live ingest
    /// because reaching `ingest_cloudevent` means standing up a whole
    /// `AppState`, a daemon and a log directory to observe one `continue`.
    ///
    /// The last case is the cross-file pin: the reserved name this filter
    /// matches has to be the one the arming probe actually sends.
    #[test]
    fn doctor_probe_events_are_not_ingested() {
        let probe = serde_json::json!({
            "hook_event_name": "PreToolUse",
            "tool_name": "OpenlatchDoctorProbe",
            "tool_input": {},
        });
        assert!(
            is_doctor_probe(Some(&probe)),
            "the probe must be skipped before process_envelope, or every doctor run \
             fabricates a tool call in the customer's audit trail"
        );

        // Real agent traffic is untouched — including a payload that merely
        // mentions the probe name somewhere other than `tool_name`.
        let real = serde_json::json!({
            "hook_event_name": "PreToolUse",
            "tool_name": "Bash",
            "tool_input": { "command": "echo OpenlatchDoctorProbe" },
        });
        assert!(!is_doctor_probe(Some(&real)));
        assert!(!is_doctor_probe(Some(&serde_json::json!({}))));
        assert!(
            !is_doctor_probe(None),
            "an envelope with no data at all is ordinary traffic, not a probe"
        );

        // The filter and the probe are in two different modules; if either side
        // renames the reserved value alone, this fails.
        assert!(is_doctor_probe(Some(&serde_json::json!({
            "tool_name": crate::hooks::bindings::codex_cli::DOCTOR_PROBE_TOOL_NAME,
        }))));
    }

    #[test]
    fn every_tool_shape_reaches_zone_mapping_with_full_input() {
        let bundle = resident(vec![], true);
        for (tool_name, tool_input) in [
            ("Read", serde_json::json!({"file_path": "/tmp/a"})),
            ("Bash", serde_json::json!({"command": ["echo a", "echo b"]})),
            (
                "mcp__storage__delete",
                serde_json::json!({"Bucket": "prod", "Keys": ["a", "b"]}),
            ),
        ] {
            let envelope: EventEnvelope = serde_json::from_value(serde_json::json!({
                "specversion": "1.0",
                "id": "evt-map",
                "source": "claude-code",
                "type": "pre_tool_use",
                "time": "2026-09-10T12:00:00Z",
                "subject": "session-map",
                "data": {
                    "tool_use_id": "tool-map",
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "cwd": "/work",
                }
            }))
            .expect("valid envelope");
            let mapped =
                map_zone_event(&envelope, &bundle, Some("agent-map")).expect("mapped event");
            assert_eq!(mapped.tool_name, tool_name);
            assert_eq!(mapped.tool_input, tool_input, "tool={tool_name}");
            assert_eq!(mapped.session_id, "session-map");
            assert_eq!(mapped.tool_use_id, "tool-map");
            assert_eq!(mapped.agent.unwrap().agent_id.as_deref(), Some("agent-map"));
        }
    }

    /// §0 site 1 — the zone evaluator reads `tool_input` / `tool_name`, and
    /// Cline sends neither. Without the fallback a Cline event arrives at the
    /// daemon, is mapped with an empty tool name and a null input, and every
    /// zone atom that reasons about either one sees nothing.
    ///
    /// Read side only: `envelope.data` is the agent's raw payload and is not
    /// rewritten — the assertions below read it back out of the envelope
    /// verbatim.
    #[test]
    fn a_cline_payload_reaches_the_zone_evaluator() {
        let bundle = resident(vec![], true);
        let envelope: EventEnvelope = serde_json::from_value(serde_json::json!({
            "specversion": "1.0",
            "id": "evt-cline",
            "source": "cline",
            "type": "pre_tool_use",
            "time": "2026-09-15T12:00:00Z",
            "subject": "session-cline",
            "data": {
                "tool_use_id": "tool-cline",
                "toolName": "run_commands",
                "parameters": { "command": "rm -rf /tmp" },
                "cwd": "/work",
            }
        }))
        .expect("valid envelope");

        let mapped = map_zone_event(&envelope, &bundle, Some("agent-cline")).expect("mapped event");

        // Evaluated in the common vocabulary: `run_commands` IS a shell call,
        // so the effect table and the shell classifier now see it as one.
        assert_eq!(mapped.tool_name, "Bash");
        assert_eq!(
            mapped.tool_input,
            serde_json::json!({ "command": "rm -rf /tmp" })
        );
        assert_eq!(mapped.session_id, "session-cline");
        assert_eq!(mapped.tool_use_id, "tool-cline");

        // The envelope itself is untouched — we read `data` differently, we do
        // not rewrite it.
        let data = envelope.data.as_ref().expect("data survives mapping");
        assert!(data.get("tool_name").is_none());
        assert!(data.get("tool_input").is_none());
        assert_eq!(
            data.get("toolName").and_then(serde_json::Value::as_str),
            Some("run_commands")
        );
    }

    /// The native spelling wins wherever an agent sends both.
    #[test]
    fn the_native_zone_spelling_wins_over_the_cline_fallback() {
        let bundle = resident(vec![], true);
        let envelope: EventEnvelope = serde_json::from_value(serde_json::json!({
            "specversion": "1.0", "id": "evt-both", "source": "claude-code",
            "type": "pre_tool_use", "time": "2026-09-15T12:00:00Z",
            "data": {
                "tool_name": "Bash",
                "toolName": "run_commands",
                "tool_input": { "command": "native" },
                "parameters": { "command": "fallback" },
            }
        }))
        .expect("valid envelope");

        let mapped = map_zone_event(&envelope, &bundle, None).expect("mapped event");

        assert_eq!(mapped.tool_name, "Bash");
        assert_eq!(
            mapped.tool_input,
            serde_json::json!({ "command": "native" })
        );
    }

    #[test]
    fn claude_native_post_payloads_reach_result_and_error_evaluation() {
        let bundle = resident(vec![], true);
        let success: EventEnvelope = serde_json::from_value(serde_json::json!({
            "specversion":"1.0","id":"evt-success","source":"claude-code",
            "type":"post_tool_use","time":"2026-09-10T12:00:00Z","subject":"session-map",
            "data":{"tool_use_id":"tool-1","tool_name":"Bash","tool_input":{"command":"cargo test"},"tool_response":"native stdout"}
        })).expect("valid native PostToolUse");
        let mapped = map_zone_event(&success, &bundle, None).expect("mapped success");
        assert_eq!(mapped.tool_result, Some(serde_json::json!("native stdout")));

        let failure: EventEnvelope = serde_json::from_value(serde_json::json!({
            "specversion":"1.0","id":"evt-failure","source":"claude-code",
            "type":"post_tool_use_failure","time":"2026-09-10T12:00:00Z","subject":"session-map",
            "data":{"tool_use_id":"tool-2","tool_name":"Bash","tool_input":{"command":"cargo test"},"error":"exit 1","is_interrupt":false}
        })).expect("valid native PostToolUseFailure");
        let mapped = map_zone_event(&failure, &bundle, None).expect("mapped failure");
        assert_eq!(mapped.event_type, "post_tool_use_failure");
        assert_eq!(
            mapped.tool_result.as_ref().and_then(|v| v.get("is_error")),
            Some(&serde_json::json!(true))
        );
        assert_eq!(
            mapped.tool_result.as_ref().and_then(|v| v.get("error")),
            Some(&serde_json::json!("exit 1"))
        );
    }

    #[test]
    fn zone_agent_type_is_the_bundle_category_not_the_agent_platform() {
        let mut bundle = resident(vec![], true);
        bundle.zone = Some(
            crate::zone_eval::load(serde_json::json!({
                "schema_version": 2,
                "organization_id": "org",
                "revision": 1,
                "built_at": "2026-09-10T00:00:00Z",
                "enforcement_enabled": true,
                "signature": null,
                "artifacts": [],
                "meta": {"agent_type": "coding"}
            }))
            .expect("zone bundle"),
        );
        let envelope: EventEnvelope = serde_json::from_value(serde_json::json!({
            "specversion": "1.0", "id": "evt", "source": "claude-code",
            "type": "pre_tool_use", "time": "2026-09-10T12:00:00Z",
            "data": {"tool_name": "Read", "tool_input": {}}
        }))
        .expect("envelope");
        let event = map_zone_event(&envelope, &bundle, Some("agent")).expect("mapped");
        assert_eq!(event.agent.unwrap().agent_type.as_deref(), Some("coding"));
    }

    #[test]
    fn a_block_with_an_unselected_rewrite_stays_blocked() {
        let mut decision = ZoneDecision {
            verdict: Verdict::Block,
            rewrite: Some(crate::zone_eval::Rewrite {
                lever: crate::generated::types::Lever("steer".into()),
                artifact_id: Some("rewrite".into()),
                steer_instruction: Some("try another way".into()),
            }),
            ..ZoneDecision::default()
        };
        assert!(!degrade_unapplied_rewrite(&mut decision));
        assert_eq!(decision.verdict, Verdict::Block);
    }

    #[test]
    fn record_v2_names_the_artifact_shadow_and_actual_delivery() {
        let mut document: serde_json::Value = serde_json::from_str(include_str!(
            "../../tools/policy-seed/fixtures/schema-2.json"
        ))
        .expect("schema-2 fixture");
        document["artifacts"][0]["policy_id"] = serde_json::json!("policy-internal");
        document["artifacts"][0]["spec_hash"] = serde_json::json!("sha256:spec");
        document["artifacts"][0]["zone_layer"] =
            serde_json::json!({"node_id": "unit-root", "path": ["unit-root"]});
        let (_, bundle) = crate::core::policy::project_document(document).expect("bundle");
        let mut monitor = ZoneDecision {
            artifact_id: Some("seed-shell".into()),
            atom_id: Some("seed-shell-atom".into()),
            would_have_verdict: Some(Verdict::Block),
            ..ZoneDecision::default()
        };
        let mut obj = serde_json::Map::new();
        stamp_record_v2(
            &mut obj,
            &monitor,
            Some(&bundle),
            "claude-code",
            Lane::Renders,
            &HookEventType::PreToolUse,
            false,
            None,
        );
        assert_eq!(obj["olpolicyruleid"], "seed-shell");
        assert_eq!(obj["olpolicyid"], "policy-internal");
        assert_eq!(obj["ollayer"], "unit-root");
        assert_eq!(obj["olspechash"], "sha256:spec");
        assert_eq!(obj["olbinding"], "claude_code_hook");
        assert_eq!(obj["olverdictshadow"], "block");
        assert_eq!(obj["olenforced"], 0);
        assert_eq!(obj["olresult"], "flagged");

        monitor.verdict = Verdict::Block;
        monitor.mode = Some(crate::generated::types::PolicyMode("enforce".into()));
        monitor.would_have_verdict = None;
        let mut post = serde_json::Map::new();
        stamp_record_v2(
            &mut post,
            &monitor,
            None,
            "claude-code",
            Lane::Renders,
            &HookEventType::PostToolUse,
            false,
            None,
        );
        assert_eq!(post["olenforced"], 0);
        assert_eq!(post["olresult"], "allowed");

        let mut pre = serde_json::Map::new();
        stamp_record_v2(
            &mut pre,
            &monitor,
            None,
            "claude-code",
            Lane::Renders,
            &HookEventType::PreToolUse,
            false,
            None,
        );
        assert_eq!(pre["olenforced"], 1);
        assert_eq!(pre["olresult"], "blocked");

        let final_block = crate::daemon::optimize::Delivery {
            enforced: Some(true),
            result: Some("blocked"),
            ..crate::daemon::optimize::Delivery::default()
        };
        let mut final_record = serde_json::Map::new();
        stamp_record_v2(
            &mut final_record,
            &monitor,
            None,
            "claude-code",
            Lane::Renders,
            &HookEventType::PreToolUse,
            false,
            Some(&final_block),
        );
        assert_eq!(final_record["olpolicyruleid"], "seed-shell");
        assert_eq!(final_record["olenforced"], 1);
        assert_eq!(final_record["olresult"], "blocked");
    }

    #[test]
    fn evaluated_monitor_ask_and_block_reach_record_v2_egress() {
        for shadow_verdict in ["ask", "block"] {
            let mut document: serde_json::Value = serde_json::from_str(include_str!(
                "../../tools/policy-seed/fixtures/schema-2.json"
            ))
            .expect("schema-2 fixture");
            document["artifacts"][0]["policy_id"] = serde_json::json!("policy-monitor");
            document["artifacts"][0]["policy_public_id"] = serde_json::json!("POL-MONITOR");
            document["artifacts"][0]["mode"] = serde_json::json!("monitor");
            document["artifacts"][0]["body"]["verdict"] = serde_json::json!(shadow_verdict);
            let (_, bundle) = crate::core::policy::project_document(document).expect("bundle");
            let envelope: EventEnvelope = serde_json::from_value(serde_json::json!({
                "specversion": "1.0", "id": format!("evt-monitor-{shadow_verdict}"),
                "source": "claude-code", "type": "pre_tool_use",
                "time": "2026-09-10T12:00:00Z", "subject": "session-monitor",
                "data": {"tool_use_id": "tool-monitor", "tool_name": "Bash",
                         "tool_input": {"command": "curl https://example.test"}}
            }))
            .expect("valid envelope");
            let event = map_zone_event(&envelope, &bundle, None).expect("mapped event");
            let (decision, _) = crate::zone_eval::evaluate(
                bundle.zone.as_ref().expect("zone bundle"),
                &event,
                None,
                1_757_505_600_000,
            );

            assert_eq!(decision.verdict, Verdict::Allow);
            assert_eq!(
                decision
                    .would_have_verdict
                    .map(|v| v.to_string())
                    .as_deref(),
                Some(shadow_verdict)
            );
            let mut obj = serde_json::Map::new();
            stamp_record_v2_if_winner(
                &mut obj,
                Some(&decision),
                None,
                Some(&bundle),
                "claude-code",
                Lane::Renders,
                &HookEventType::PreToolUse,
                false,
                None,
            );

            assert_eq!(obj["olatomid"], "seed-shell-atom");
            assert_eq!(obj["olpolicyruleid"], "seed-shell");
            assert_eq!(obj["olpolicyid"], "policy-monitor");
            assert_eq!(obj["olverdictshadow"], shadow_verdict);
            assert_eq!(obj["olenforced"], 0);
            assert_eq!(obj["olresult"], "flagged");
        }
    }

    #[test]
    fn enforcing_legacy_match_keeps_its_provenance_over_a_zone_decision() {
        let bundle = resident(
            vec![rule("OL-LEGACY-WINNER", "*rm*", PolicyRuleMode::Enforce)],
            true,
        );
        let legacy = match_of("OL-LEGACY-WINNER", false);
        let zone = ZoneDecision {
            verdict: Verdict::Allow,
            artifact_id: Some("zone-loser".into()),
            atom_id: Some("zone-atom".into()),
            ..ZoneDecision::default()
        };
        let mut obj = serde_json::Map::new();
        stamp_policy_extensions(
            &mut obj,
            Some(&legacy),
            Some(&bundle),
            false,
            bundle.built_at,
        );
        stamp_record_v2_if_winner(
            &mut obj,
            Some(&zone),
            Some(&legacy),
            Some(&bundle),
            "claude-code",
            Lane::Renders,
            &HookEventType::PreToolUse,
            false,
            None,
        );

        assert_eq!(obj["olpolicyruleid"], "OL-LEGACY-WINNER");
        assert!(obj.get("olatomid").is_none());
        assert!(obj.get("olresult").is_none());
        assert!(obj.get("olenforced").is_none());
    }

    fn match_of(rule_id: &str, shadow: bool) -> PolicyMatch {
        PolicyMatch {
            rule_id: rule_id.to_string(),
            reason: format!("{rule_id} says no"),
            severity: PolicyRuleSeverity::High,
            mode: if shadow {
                PolicyRuleMode::Observe
            } else {
                PolicyRuleMode::Enforce
            },
            shadow,
        }
    }

    fn outcome(verdict: Verdict, id: &str) -> BatchOutcome {
        BatchOutcome {
            verdict,
            event_type: match verdict {
                Verdict::Ask => HookEventType::Stop,
                _ => HookEventType::PreToolUse,
            },
            event_id: id.to_string(),
            policy_match: match verdict {
                Verdict::Block => Some(match_of("OL-CMD-ENF", false)),
                _ => None,
            },
            zone_decision: None,
            hold_wait: None,
            delivery: crate::daemon::optimize::Delivery::default(),
        }
    }

    // -- per-envelope verdict -----------------------------------------------

    #[test]
    fn a_non_shadow_match_blocks_and_a_shadow_match_does_not() {
        assert_eq!(
            verdict_for(
                &HookEventType::PreToolUse,
                None,
                Some(&match_of("OL-CMD-A", false))
            ),
            Verdict::Block
        );
        assert_eq!(
            verdict_for(
                &HookEventType::PreToolUse,
                None,
                Some(&match_of("OL-CMD-A", true))
            ),
            Verdict::Allow,
            "observe never blocks"
        );
        assert_eq!(
            verdict_for(&HookEventType::PreToolUse, None, None),
            Verdict::Allow
        );
        // D14 correction: `Stop` returns `allow`, never `ask`. See `verdict_for`.
        assert_eq!(
            verdict_for(&HookEventType::Stop, None, None),
            Verdict::Allow
        );
    }

    /// A block wins on a Stop envelope too — nothing about the event type
    /// softens a non-shadow match.
    #[test]
    fn block_outranks_the_stop_arm() {
        assert_eq!(
            verdict_for(
                &HookEventType::Stop,
                None,
                Some(&match_of("OL-CMD-A", false))
            ),
            Verdict::Block
        );
    }

    // -- the batch lattice (4b) ---------------------------------------------

    #[test]
    fn the_fold_is_most_restrictive_wins() {
        for (a, b) in [
            (Verdict::Allow, Verdict::Optimize),
            (Verdict::Allow, Verdict::Ask),
            (Verdict::Allow, Verdict::Block),
            (Verdict::Optimize, Verdict::Ask),
            (Verdict::Optimize, Verdict::Block),
            (Verdict::Ask, Verdict::Block),
        ] {
            for (first, second) in [(a, b), (b, a)] {
                let mut worst = None;
                join_most_restrictive(&mut worst, outcome(first, "evt_1"));
                join_most_restrictive(&mut worst, outcome(second, "evt_2"));
                let got = worst.expect("folded");
                assert_eq!(
                    verdict_rank(got.verdict),
                    verdict_rank(a).max(verdict_rank(b)),
                    "{first:?} then {second:?} must fold to the more restrictive"
                );
            }
        }
    }

    /// Order-independence over a whole batch: the folded verdict must not
    /// depend on where the denying envelope sits.
    #[test]
    fn the_fold_is_order_independent() {
        let orders = [
            [Verdict::Block, Verdict::Ask, Verdict::Allow],
            [Verdict::Allow, Verdict::Block, Verdict::Ask],
            [Verdict::Ask, Verdict::Allow, Verdict::Block],
        ];
        for order in orders {
            let mut worst = None;
            for (i, v) in order.iter().enumerate() {
                join_most_restrictive(&mut worst, outcome(*v, &format!("evt_{i}")));
            }
            assert_eq!(worst.expect("folded").verdict, Verdict::Block, "{order:?}");
        }
    }

    /// Within one rank the earliest envelope wins — the same tiebreak the
    /// pre-policy fold had, so the reported `event_id` stays deterministic.
    #[test]
    fn equal_rank_keeps_the_first_envelope() {
        let mut worst = None;
        join_most_restrictive(&mut worst, outcome(Verdict::Allow, "evt_1"));
        join_most_restrictive(&mut worst, outcome(Verdict::Allow, "evt_2"));
        assert_eq!(worst.expect("folded").event_id, "evt_1");
    }

    /// The deciding envelope's id and match travel together, so the response's
    /// `reason` / `rule_id` describe the block the agent is being told about.
    #[test]
    fn the_deciding_envelope_carries_its_own_match() {
        let mut worst = None;
        join_most_restrictive(&mut worst, outcome(Verdict::Allow, "evt_1"));
        join_most_restrictive(&mut worst, outcome(Verdict::Block, "evt_2"));
        let got = worst.expect("folded");
        assert_eq!(got.event_id, "evt_2");
        assert_eq!(
            got.policy_match.expect("block carries its match").rule_id,
            "OL-CMD-ENF"
        );
    }

    // -- the six extensions (4a / D15) --------------------------------------

    fn stamped(
        policy_match: Option<&PolicyMatch>,
        bundle: Option<&ResidentBundle>,
        offline: bool,
        now: SystemTime,
    ) -> serde_json::Map<String, serde_json::Value> {
        let mut obj = serde_json::Map::new();
        stamp_policy_extensions(&mut obj, policy_match, bundle, offline, now);
        obj
    }

    #[test]
    fn a_deny_stamps_rule_id_revision_age_and_offline_but_no_shadow() {
        let bundle = resident(
            vec![rule("OL-CMD-ENF", "*rm*", PolicyRuleMode::Enforce)],
            true,
        );
        let now = bundle.built_at + Duration::from_secs(90);
        let obj = stamped(
            Some(&match_of("OL-CMD-ENF", false)),
            Some(&bundle),
            false,
            now,
        );

        assert_eq!(obj.get("olpolicyruleid").unwrap(), "OL-CMD-ENF");
        assert_eq!(obj.get("olpolicybundlerev").unwrap(), 42);
        assert_eq!(obj.get("olpolicybundleage").unwrap(), 90);
        assert_eq!(obj.get("olpolicyoffline").unwrap(), false);
        assert_eq!(
            obj.get("olcontractranges").unwrap(),
            "decision_event=1-2,policy_bundle=1-2"
        );
        assert_eq!(obj.get("olcompatversion").unwrap(), 1);
        assert_eq!(obj.get("olresidentbundleschema").unwrap(), 1);
        assert!(obj.get("olcompatobservedat").is_some());
        assert!(
            obj.get("olverdictshadow").is_none(),
            "an enforced deny is not a shadow verdict"
        );
    }

    #[test]
    fn an_observe_match_stamps_the_shadow_verdict() {
        let bundle = resident(vec![], true);
        let obj = stamped(
            Some(&match_of("OL-CMD-OBS", true)),
            Some(&bundle),
            false,
            bundle.built_at,
        );
        // D14: the shadow verdict is stamped with the frozen token, not `deny`.
        assert_eq!(obj.get("olverdictshadow").unwrap(), "block");
        assert_eq!(obj.get("olpolicyruleid").unwrap(), "OL-CMD-OBS");
    }

    /// D29 / PRD "Empty `rules` array": a resident bundle with nothing in it
    /// still reports its revision, which is what distinguishes "allowed because
    /// the org has no rules" from "allowed because we have no bundle".
    #[test]
    fn no_match_still_reports_the_bundle_revision() {
        let bundle = resident(vec![], true);
        let obj = stamped(None, Some(&bundle), false, bundle.built_at);
        assert_eq!(obj.get("olpolicybundlerev").unwrap(), 42);
        assert!(obj.get("olpolicyruleid").is_none());
        assert!(obj.get("olverdictshadow").is_none());
    }

    /// PRD "No bundle ever fetched": the bundle fields are OMITTED, not zeroed.
    #[test]
    fn with_no_bundle_the_bundle_fields_are_omitted_and_offline_still_reported() {
        let obj = stamped(None, None, true, SystemTime::now());
        assert!(obj.get("olpolicybundlerev").is_none());
        assert!(obj.get("olresidentbundleschema").is_none());
        assert!(obj.get("olresidentbundledigest").is_none());
        assert!(obj.get("olpolicybundleage").is_none());
        assert_eq!(obj.get("olpolicyoffline").unwrap(), true);
    }

    /// Clock skew must never yield a negative age on the wire.
    #[test]
    fn bundle_age_is_clamped_at_zero_when_the_local_clock_is_behind() {
        let bundle = resident(vec![], true);
        let behind = bundle.built_at - Duration::from_secs(3_600);
        let obj = stamped(None, Some(&bundle), false, behind);
        assert_eq!(obj.get("olpolicybundleage").unwrap(), 0);
    }

    // -- /metrics (D48, D29) ------------------------------------------------

    fn runtime(
        bundle: Option<ResidentBundle>,
        last_poll_ok_at: i64,
    ) -> crate::daemon::PolicyRuntime {
        let has_bundle = bundle.is_some();
        crate::daemon::PolicyRuntime {
            handle: crate::core::policy::new_handle(bundle),
            last_fetch_ok: Arc::new(std::sync::atomic::AtomicBool::new(true)),
            last_poll_ok_at: Arc::new(std::sync::atomic::AtomicI64::new(last_poll_ok_at)),
            last_download_verified_at: Arc::new(std::sync::atomic::AtomicI64::new(0)),
            readiness: Arc::new(std::sync::RwLock::new(crate::daemon::PolicyReadiness {
                phase: if has_bundle {
                    crate::daemon::PolicyReadinessPhase::Resident
                } else {
                    crate::daemon::PolicyReadinessPhase::Waiting
                },
                last_attempted_at: None,
                last_outcome: None,
                reason_code: None,
                selection_acknowledged: false,
                compatibility_diagnostic_valid: false,
            })),
            floor: Arc::new(std::sync::RwLock::new(None)),
            sessions: Arc::new(crate::daemon::session_state::SessionStateManager::default()),
            dispatch_sessions: Arc::new(
                crate::daemon::session_state::DispatchStateManager::default(),
            ),
            holds: Arc::new(crate::daemon::hold_queue::HoldQueue::default()),
            reinforcement: Arc::new(crate::daemon::reinforcement::ReinforcementStore::default()),
        }
    }

    fn schema_two_resident(artifacts: serde_json::Value) -> ResidentBundle {
        let document = serde_json::json!({
            "schema_version": 2,
            "organization_id": "org",
            "revision": 2,
            "built_at": "2026-09-01T00:00:00Z",
            "enforcement_enabled": true,
            "rules": [],
            "signature": null,
            "artifacts": artifacts,
        });
        crate::core::policy::project_document(document)
            .expect("schema-2 fixture projects")
            .1
    }

    fn t1_artifact(
        id: &str,
        policy_id: Option<&str>,
        public_id: Option<&str>,
    ) -> serde_json::Value {
        serde_json::json!({
            "artifact_id": id,
            "atom_id": format!("atom-{id}"),
            "policy_id": policy_id,
            "policy_public_id": public_id,
            "kind": "t1_predicate_tree",
            "tier": 1,
            "mode": "enforce",
            "body": {
                "node": {"op": "leaf", "leaf": {"pred": "equals", "field": "tool.name", "value": "Bash"}},
                "verdict": "block",
                "reason": "test"
            }
        })
    }

    // --- health ------------------------------------------------------------

    #[test]
    fn health_degrades_on_a_failed_egress_even_with_every_subsystem_running() {
        assert_eq!(health_status(false, false), "ok");
        assert_eq!(
            health_status(false, true),
            "degraded",
            "a daemon that cannot get a byte out is not healthy"
        );
        assert_eq!(health_status(true, false), "degraded");
        assert_eq!(health_status(true, true), "degraded");
    }

    // --- egress metrics ---------------------------------------------------

    fn proxied_egress() -> crate::egress::EgressConfig {
        let toml = crate::egress::ProxyToml {
            mode: Some("manual".into()),
            url: Some("http://alice-proxy.corp:8080".into()),
            ..Default::default()
        };
        struct NoEnv;
        impl crate::egress::EnvSource for NoEnv {
            fn var(&self, _key: &str) -> Option<String> {
                None
            }
        }
        crate::egress::EgressConfig::resolve(Some(&toml), &NoEnv, 7443, 7444).expect("resolve")
    }

    #[test]
    fn egress_metrics_report_a_healthy_direct_host_as_up() {
        let m = egress_metrics(&crate::egress::EgressState::new(
            &crate::egress::EgressConfig::direct(),
        ));
        assert!(m.up);
        assert_eq!(m.status, "ok");
        assert_eq!(m.consecutive_failures, 0);
        assert!(!m.proxy_in_use);
    }

    #[test]
    fn egress_metrics_go_down_only_at_the_threshold() {
        let state = crate::egress::EgressState::new(&crate::egress::EgressConfig::direct());
        state.record_failure(crate::error::ERR_PROXY_UNREACHABLE, "refused");
        let one = egress_metrics(&state);
        assert!(one.up, "one blip is not an outage");
        assert_eq!(one.consecutive_failures, 1);

        state.record_failure(crate::error::ERR_PROXY_UNREACHABLE, "refused");
        let two = egress_metrics(&state);
        assert!(!two.up);
        assert_eq!(two.status, "failed");
        assert_eq!(two.consecutive_failures, 2);
    }

    /// The privacy invariant, asserted on the rendered JSON rather than argued
    /// in a comment: `/metrics` is unauthenticated, so the proxy host must not
    /// appear anywhere in it.
    #[test]
    fn the_unauthenticated_scrape_never_carries_the_proxy_url() {
        let cfg = proxied_egress();
        let state = crate::egress::EgressState::new(&cfg);
        let m = egress_metrics(&state);
        assert!(m.proxy_in_use);

        let rendered = serde_json::json!({
            "egress_up": m.up,
            "egress_status": m.status,
            "egress_consecutive_failures": m.consecutive_failures,
            "proxy_in_use": m.proxy_in_use,
        })
        .to_string();
        assert!(
            !rendered.contains("alice-proxy.corp"),
            "the proxy host leaked into an unauthenticated surface: {rendered}"
        );
        assert!(!rendered.contains("8080"));
    }

    // --- egress shape stamping (D-29, I-4) --------------------------------

    /// Apply a stamp to a minimal hook-derived envelope and return the wire bytes.
    ///
    /// Asserted on the serialized JSON and nowhere else: the three are undeclared
    /// extension attributes with no typed field to inspect, so the wire bytes are the
    /// only place the contract exists — in particular that an absent attribute is
    /// *absent*, not `null`.
    fn stamped_envelope(snapshot: &crate::egress::EgressSnapshot) -> serde_json::Value {
        let envelope = EventEnvelope {
            specversion: "1.0".into(),
            id: new_event_id(),
            source: AgentType::ClaudeCode,
            type_: HookEventType::PreToolUse,
            time: chrono::Utc::now(),
            datacontenttype: None,
            subject: Some("sess_abc".into()),
            data: None,
            os: None,
            arch: None,
            localipv4: None,
            localipv6: None,
            publicipv4: None,
            publicipv6: None,
            clientversion: None,
            agentversion: None,
            agentid: None,
            osuser: None,
            gitemail: None,
            provideracct: None,
            wireformat: None,
            hostid: None,
            hostname: None,
        };
        let mut wire = serde_json::to_value(&envelope).expect("serialise envelope");
        stamp_egress_extensions(
            wire.as_object_mut().expect("an envelope is a JSON object"),
            egress_stamp(snapshot),
        );
        wire
    }

    fn snapshot_of(cfg: &crate::egress::EgressConfig) -> crate::egress::EgressSnapshot {
        crate::egress::EgressSnapshot::from_config(cfg)
    }

    /// Direct: all three absent. Absence is the platform's "no chip" signal, so a `null`
    /// or a `"direct"` here would draw a chip for a host that has no proxy at all.
    #[test]
    fn a_direct_host_stamps_none_of_the_three() {
        let wire = stamped_envelope(&snapshot_of(&crate::egress::EgressConfig::direct()));
        for attribute in ["proxytype", "proxysource", "proxyintercepted"] {
            assert!(
                wire.get(attribute).is_none(),
                "{attribute} must be ABSENT on a direct host, not null: {wire}"
            );
        }
    }

    /// Proxied, no TLS observation yet: type and source ride, interception does not.
    #[test]
    fn a_proxied_host_stamps_type_and_source_but_not_interception() {
        let wire = stamped_envelope(&snapshot_of(&proxied_egress()));
        assert_eq!(wire["proxytype"], "http");
        assert!(
            wire.get("proxyintercepted").is_none(),
            "absence must mean 'not yet known', never 'not intercepted': {wire}"
        );
        // And the address never rides along, even toward our own platform (D-29).
        let rendered = wire.to_string();
        assert!(!rendered.contains("alice-proxy.corp"), "{rendered}");
        assert!(!rendered.contains("8080"), "{rendered}");
    }

    /// Once a handshake has been observed, the verdict rides — including `false`, which is
    /// a real observation and not the same thing as absence.
    #[test]
    fn an_observed_handshake_stamps_the_interception_verdict() {
        let mut snapshot = snapshot_of(&proxied_egress());
        snapshot.source = Some(crate::egress::ProxySource::Windows);
        snapshot.tls_intercepted = Some(true);
        let wire = stamped_envelope(&snapshot);
        assert_eq!(wire["proxytype"], "http");
        assert_eq!(wire["proxysource"], "windows");
        assert_eq!(wire["proxyintercepted"], true);

        snapshot.tls_intercepted = Some(false);
        assert_eq!(stamped_envelope(&snapshot)["proxyintercepted"], false);
    }

    /// A PAC-sourced route reports `pac` whatever scheme the file returned — the frozen
    /// contract's sharpest edge, and the one an operator debugging PAC needs.
    #[test]
    fn a_pac_route_stamps_pac_not_the_returned_scheme() {
        let mut snapshot = snapshot_of(&proxied_egress());
        snapshot.source = Some(crate::egress::ProxySource::Pac);
        let wire = stamped_envelope(&snapshot);
        assert_eq!(wire["proxytype"], "pac");
        assert_eq!(wire["proxysource"], "pac");
    }

    /// A proxy whose provenance was never recorded still reports its type. `proxysource`
    /// is nullable in the config, so stamping a guess would be inventing one.
    #[test]
    fn a_proxy_with_no_recorded_source_still_stamps_its_type() {
        let wire = stamped_envelope(&snapshot_of(&proxied_egress()));
        assert_eq!(wire["proxytype"], "http");
        assert!(wire.get("proxysource").is_none());
    }

    #[test]
    fn metrics_report_policy_disabled_as_such() {
        let m = policy_metrics(None, SystemTime::now());
        assert!(!m.enabled);
        assert!(!m.has_bundle);
        assert!(m.schema_version.is_null());
        assert!(m.bundle_digest.is_null());
        assert!(m.revision.is_null());
        assert!(m.bundle_age_seconds.is_null());
        assert!(m.last_poll_ok_secs.is_null());
    }

    /// **D29.** "Enabled but never fetched a bundle" — the cold-start case with
    /// no last-good to fall back to — must be visibly different from "enabled
    /// and enforcing". Conflating them is how a client silently disarms on
    /// first boot and nobody notices.
    #[test]
    fn metrics_distinguish_no_bundle_from_a_resident_one() {
        let cold = policy_metrics(Some(&runtime(None, 0)), SystemTime::now());
        assert!(cold.enabled);
        assert!(!cold.has_bundle, "never fetched: allow, marked no-bundle");
        assert!(cold.schema_version.is_null());
        assert!(cold.bundle_digest.is_null());
        assert!(cold.revision.is_null());
        assert!(cold.last_poll_ok_secs.is_null(), "0 means never, not now");
        assert!(cold.last_platform_contact_at.is_null());
        assert!(cold.last_download_verified_at.is_null());

        let bundle = resident(vec![], true);
        let now = bundle.built_at + Duration::from_secs(120);
        let warm = policy_metrics(Some(&runtime(Some(bundle), 1_784_624_400)), now);
        assert!(warm.has_bundle, "last-good resident: keep enforcing");
        assert_eq!(warm.schema_version, 1);
        assert!(warm.bundle_digest.is_null());
        assert_eq!(warm.revision, 42);
        assert_eq!(warm.bundle_age_seconds, 120);
        assert!(warm.resident_aip_count.is_null());
    }

    #[test]
    fn metrics_count_distinct_loaded_aips_and_exclude_exceptions() {
        let mut exception = t1_artifact("exception", Some("policy-exception"), Some("AIP-999"));
        exception["kind"] = serde_json::json!("exception");
        exception["body"] = serde_json::json!({
            "scope": "exact_action",
            "selector": {"action_hash": "abc"},
            "verdict": "allow"
        });
        let bundle = schema_two_resident(serde_json::json!([
            t1_artifact("a", Some("policy-2"), Some("AIP-002")),
            t1_artifact("b", Some("policy-2"), Some("AIP-002")),
            t1_artifact("c", Some("policy-3"), Some("AIP-003")),
            exception,
            {"artifact_id": "skipped", "policy_id": "policy-skipped", "kind": "future_kind", "body": {}}
        ]));
        let metrics = policy_metrics(Some(&runtime(Some(bundle), 1_000)), SystemTime::now());
        assert_eq!(metrics.resident_aip_count, 2);
        assert_eq!(metrics.skipped_items, 1);
    }

    #[test]
    fn metrics_report_zero_for_an_empty_schema_two_bundle() {
        let bundle = schema_two_resident(serde_json::json!([]));
        let metrics = policy_metrics(Some(&runtime(Some(bundle), 1_000)), SystemTime::now());
        assert_eq!(metrics.resident_aip_count, 0);
    }

    #[test]
    fn metrics_require_a_nonempty_canonical_policy_id_on_every_loaded_artifact() {
        for artifact in [
            t1_artifact("missing", None, Some("AIP-002")),
            t1_artifact("empty", Some(""), Some("AIP-002")),
        ] {
            let bundle = schema_two_resident(serde_json::json!([artifact]));
            let metrics = policy_metrics(Some(&runtime(Some(bundle), 1_000)), SystemTime::now());
            assert!(metrics.resident_aip_count.is_null());
        }
    }

    /// Which agent the platform thinks this install is — the field that says
    /// whether a rule carrying `conditions` can fire here at all. Null and a
    /// value must be distinguishable, or "the rule did not match" and "the
    /// context never arrived" are the same reading from outside the process.
    #[test]
    fn metrics_report_the_resident_agent_function() {
        let mut bundle = resident(vec![], true);
        bundle.agent_function = Some(crate::generated::types::AgentFunction::ItOps);
        let m = policy_metrics(Some(&runtime(Some(bundle), 1_000)), SystemTime::now());
        assert_eq!(
            m.agent_function, "it_ops",
            "the wire spelling, not the variant"
        );

        // No context is null, not a string — every way of not having one
        // (no client_config, no agent_context, an unparsable value) lands here.
        let contextless = policy_metrics(
            Some(&runtime(Some(resident(vec![], true)), 1_000)),
            SystemTime::now(),
        );
        assert!(contextless.agent_function.is_null());
        // And so is having no bundle at all.
        assert!(policy_metrics(Some(&runtime(None, 0)), SystemTime::now())
            .agent_function
            .is_null());
    }

    #[test]
    fn metrics_bundle_age_is_clamped_and_poll_age_is_seconds_since_the_last_ok() {
        let bundle = resident(vec![], true);
        let behind = bundle.built_at - Duration::from_secs(60);
        let m = policy_metrics(Some(&runtime(Some(bundle), 1_000)), behind);
        assert_eq!(m.bundle_age_seconds, 0, "clock skew never goes negative");
        let expected = behind
            .duration_since(UNIX_EPOCH)
            .expect("after epoch")
            .as_secs() as i64
            - 1_000;
        assert_eq!(m.last_poll_ok_secs, expected);
    }

    #[test]
    fn metrics_keep_verified_receipt_separate_from_platform_contact_and_readiness() {
        let runtime = runtime(Some(resident(vec![], true)), 1_784_624_500);
        runtime
            .last_download_verified_at
            .store(1_784_624_400, Ordering::Relaxed);
        *runtime.readiness.write().unwrap() = crate::daemon::PolicyReadiness {
            phase: crate::daemon::PolicyReadinessPhase::Retrying,
            last_attempted_at: Some(1_784_624_500),
            last_outcome: Some("http_202"),
            reason_code: Some("install_materializing"),
            selection_acknowledged: true,
            compatibility_diagnostic_valid: false,
        };
        let m = policy_metrics(
            Some(&runtime),
            UNIX_EPOCH + Duration::from_secs(1_784_624_520),
        );
        assert_eq!(m.last_download_verified_at, "2026-07-21T09:00:00Z");
        assert_eq!(m.last_platform_contact_at, "2026-07-21T09:01:40Z");
        assert_eq!(m.readiness_state, "retrying");
        assert_eq!(m.last_outcome, "http_202");
        assert_eq!(m.readiness_reason, "install_materializing");
    }

    #[test]
    fn nonpositive_receipt_timestamps_are_unknown_not_epoch_dates() {
        for receipt in [0, -1] {
            let runtime = runtime(Some(resident(vec![], true)), 0);
            runtime
                .last_download_verified_at
                .store(receipt, Ordering::Relaxed);

            let metrics = policy_metrics(Some(&runtime), SystemTime::now());

            assert!(metrics.last_download_verified_at.is_null());
        }
    }

    fn compatibility_evidence() -> crate::core::protocol::contracts::CompatibilityState {
        use crate::core::protocol::contracts::{
            CompatibilityDiagnostic, CompatibilitySelection, CompatibilityState, VersionRange,
            POLICY_BUNDLE_FAMILY,
        };

        let mut state = CompatibilityState::default();
        state.selections.insert(
            POLICY_BUNDLE_FAMILY.to_string(),
            CompatibilitySelection {
                version: 2,
                selected_at: "2026-09-28T10:00:00Z".to_string(),
            },
        );
        state.selections.insert(
            crate::core::protocol::contracts::DECISION_EVENT_FAMILY.to_string(),
            CompatibilitySelection {
                version: 2,
                selected_at: "2026-09-28T10:00:00Z".to_string(),
            },
        );
        state.diagnostics.insert(
            POLICY_BUNDLE_FAMILY.to_string(),
            CompatibilityDiagnostic {
                family: POLICY_BUNDLE_FAMILY.to_string(),
                client_range: VersionRange::new(1, 2).unwrap(),
                platform_range: VersionRange::new(1, 2),
                last_selection: Some(2),
                observed_at: "2026-09-28T10:01:00Z".to_string(),
                detail: "temporarily unservable".to_string(),
            },
        );
        state
    }

    #[test]
    fn pending_keeps_selection_while_legacy_no_ack_stays_invalid_after_transport() {
        use crate::core::protocol::contracts::{DECISION_EVENT_FAMILY, POLICY_BUNDLE_FAMILY};

        let runtime = runtime(Some(resident(vec![], true)), 0);
        {
            let mut readiness = runtime.readiness.write().unwrap();
            readiness.selection_acknowledged = true;
            readiness.compatibility_diagnostic_valid = false;
            readiness.last_outcome = Some("http_202");
        }
        let mut pending = compatibility_evidence();
        mask_stale_policy_compatibility(&runtime, &mut pending);
        assert!(pending.selections.contains_key(POLICY_BUNDLE_FAMILY));
        assert!(pending.selections.contains_key(DECISION_EVENT_FAMILY));
        assert!(!pending.diagnostics.contains_key(POLICY_BUNDLE_FAMILY));

        {
            let mut readiness = runtime.readiness.write().unwrap();
            readiness.selection_acknowledged = false;
            readiness.last_outcome = Some("transport_error");
            readiness.reason_code = Some("transport_error");
        }
        let mut legacy_then_transport = compatibility_evidence();
        mask_stale_policy_compatibility(&runtime, &mut legacy_then_transport);
        assert!(!legacy_then_transport
            .selections
            .contains_key(POLICY_BUNDLE_FAMILY));
        assert!(legacy_then_transport
            .selections
            .contains_key(DECISION_EVENT_FAMILY));
        assert!(!legacy_then_transport
            .diagnostics
            .contains_key(POLICY_BUNDLE_FAMILY));
    }

    // -- deny vs. a queued config alert (4c) --------------------------------

    fn queued_alert() -> crate::daemon::config_monitor::PendingAlert {
        crate::daemon::config_monitor::PendingAlert {
            alert_id: "alert_1".to_string(),
            session_ref_id: "sess_abc".to_string(),
            severity: "high".to_string(),
            headline: "Configuration alert pending".to_string(),
            body: "MCP server 'evil' was added.".to_string(),
            source_id: "claude-code:mcp:deadbeef".to_string(),
            created_at: "2026-07-21T09:00:00Z".to_string(),
            delivered: false,
        }
    }

    /// The collision the 4c guard exists to prevent, demonstrated in both
    /// directions: a policy block names its rule, and `attach_alert_context`
    /// would have overwritten that with the alert's text — attributing the
    /// block to the wrong thing. Hence the alert is skipped (and left queued)
    /// whenever the folded verdict is a block.
    #[test]
    fn attaching_an_alert_to_a_block_would_overwrite_the_rules_reason() {
        let m = match_of("OL-CMD-ENF", false);
        let mut response = VerdictResponse::block(
            "evt_1".to_string(),
            1.0,
            Some(m.reason.clone()),
            Some(m.severity.to_string()),
            Some(m.rule_id.clone()),
        );
        assert_eq!(response.reason.as_deref(), Some("OL-CMD-ENF says no"));
        assert!(response.context.is_none());

        attach_alert_context(&mut response, &queued_alert());

        assert_eq!(
            response.reason.as_deref(),
            Some("Configuration alert pending"),
            "attach_alert_context overwrites `reason` unconditionally"
        );
        assert!(
            response.context.is_some(),
            "and the translator prefers context over reason on a block"
        );
    }

    /// The rendering half: with the guard in place the hook binary shows the
    /// developer the rule's own reason, which is the "every blocked action
    /// names the rule that blocked it" user story.
    #[test]
    fn a_guarded_block_renders_the_rules_reason_to_the_agent() {
        let m = match_of("OL-CMD-ENF", false);
        let response = VerdictResponse::block(
            "evt_1".to_string(),
            1.0,
            Some(m.reason.clone()),
            Some(m.severity.to_string()),
            Some(m.rule_id.clone()),
        );
        let out = crate::hook_output::translate(
            "claude-code",
            "pre_tool_use",
            &crate::hook_output::Verdict {
                decision: &response.verdict.to_string(),
                reason: response.reason.as_deref(),
                context: None,
            },
        );
        assert_eq!(
            out["hookSpecificOutput"]["permissionDecision"], "deny",
            "a policy block must reach the agent as a native deny"
        );
        assert_eq!(
            out["hookSpecificOutput"]["permissionDecisionReason"],
            "OL-CMD-ENF says no"
        );
    }

    /// Every stamped name must be a legal CloudEvents extension attribute
    /// (`^[a-z0-9]+$`, `.claude/rules/envelope-format.md`).
    #[test]
    fn extension_names_are_lowercase_alphanumeric() {
        let bundle = resident(vec![], true);
        let obj = stamped(
            Some(&match_of("OL-CMD-A", true)),
            Some(&bundle),
            true,
            bundle.built_at,
        );
        assert_eq!(
            obj.len(),
            9,
            "five policy extensions plus four compatibility facts; olverdict is stamped by the caller"
        );
        for name in obj.keys() {
            assert!(
                name.chars()
                    .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit()),
                "illegal CloudEvents extension name: {name}"
            );
        }
    }

    // -- identity stamping gate (I-1 D-14) -----------------------------------

    /// Resolvers with values that could only have come from the injected
    /// closures. Nothing here touches `identity::os_user` or the process-global
    /// session cache: the real `os_user` memoises its first answer for the whole
    /// test binary, so reaching it through the env seam would both freeze that
    /// value and race the `os_user` precedence test.
    /// Map one payload and read back the `data` that would travel.
    fn common_of(data: serde_json::Value, event_type: HookEventType) -> serde_json::Value {
        let mut obj = serde_json::Map::new();
        obj.insert("data".to_string(), data);
        map_data_to_common(&mut obj, &event_type);
        obj.remove("data").expect("data survives the mapping")
    }

    /// A replayed event is still a normalized event.
    ///
    /// The fail-open path writes the envelope to `fallback.jsonl` and the
    /// daemon replays it later. That body is rebuilt from the raw envelope so
    /// it cannot claim a verdict or a bundle it never saw — and rebuilding it
    /// used to drop the agent facts along with them, which are read from
    /// `data` and do not go stale. For an agent that nests its hook fields
    /// nothing downstream could recover them, so every event that failed open
    /// was also permanently unnormalized.
    #[test]
    fn a_replayed_envelope_is_still_mapped_and_claims_no_verdict() {
        let envelope = EventEnvelope {
            specversion: "1.0".into(),
            id: new_event_id(),
            source: AgentType::Cline,
            type_: HookEventType::PreToolUse,
            time: chrono::Utc::now(),
            datacontenttype: None,
            subject: Some("sess-replayed".into()),
            data: Some(serde_json::json!({
                "preToolUse": {
                    "toolName": "execute_command",
                    "parameters": {"command": "npm test"},
                    "workspaceRoots": ["/work/scratch"]
                }
            })),
            os: None,
            arch: None,
            localipv4: None,
            localipv6: None,
            publicipv4: None,
            publicipv6: None,
            clientversion: None,
            agentversion: None,
            agentid: None,
            osuser: None,
            gitemail: None,
            provideracct: None,
            wireformat: None,
            hostid: None,
            hostname: None,
        };

        let body = replay_cloud_body(&envelope, &HookEventType::PreToolUse);

        // The payload is mapped, read out of the NESTED shape.
        assert_eq!(body["data"]["tool_name"], "Bash");
        assert_eq!(body["data"]["cwd"], "/work/scratch");
        assert_eq!(body["data"]["tool_input"]["command"], "npm test");

        // The client's state at replay time does not.
        for stale in [
            "olverdict",
            "ollatencyms",
            "olpolicybundlerev",
            "olpolicybundleage",
            "olpolicyoffline",
        ] {
            assert!(
                body.get(stale).is_none(),
                "{stale} describes the client NOW, not this replayed event"
            );
        }

        // And the agent's own bytes ride verbatim alongside the mapping.
        assert_eq!(
            body["data"]["raw_agent_payload"]["preToolUse"]["toolName"], "execute_command",
            "the original must survive a replay too"
        );
    }

    /// **The whole point of mapping.** One tool call, described three ways by
    /// three agents, travelling as one shape.
    ///
    /// The payload shapes are the ones the agents actually send: Claude Code
    /// and Codex flat and snake_case, Cline nested under its hook's own name.
    /// A consumer reading `tool_name` never has to know which it received.
    #[test]
    fn three_agents_one_shape() {
        let native = serde_json::json!({
            "session_id": "sess_a",
            "cwd": "/repo",
            "hook_event_name": "PreToolUse",
            "tool_name": "Bash",
            "tool_input": {"command": "cargo test"},
            "tool_use_id": "toolu_01"
        });
        let claude = common_of(native.clone(), HookEventType::PreToolUse);
        let codex = common_of(
            serde_json::json!({
                "session_id": "019a5f3e",
                "turn_id": "019a5f41",
                "cwd": "/repo",
                "hook_event_name": "PreToolUse",
                "model": "gpt-5.4",
                "tool_name": "Bash",
                "tool_input": {"command": "cargo test"},
                "tool_use_id": "call_Zq8"
            }),
            HookEventType::PreToolUse,
        );
        let cline = common_of(
            serde_json::json!({
                "taskId": "conv_1",
                "hookName": "PreToolUse",
                "clineVersion": "4.1.17",
                "workspaceRoots": ["/repo"],
                "model": {"provider": "gemini", "slug": "gemini-3.1-flash-lite"},
                "preToolUse": {
                    "toolName": "execute_command",
                    "parameters": {"command": "cargo test"}
                }
            }),
            HookEventType::PreToolUse,
        );

        // The same three questions, answered off one key each, for all three.
        for d in [&claude, &codex, &cline] {
            assert_eq!(d["cwd"], "/repo", "{d}");
            assert_eq!(d["hook_event_name"], "PreToolUse", "{d}");
            assert!(d["tool_name"].is_string(), "{d}");
            assert_eq!(d["tool_input"]["command"], "cargo test", "{d}");
        }
        assert_eq!(claude["tool_name"], "Bash");
        // One vocabulary, too: Cline's `execute_command` travels as `Bash`.
        assert_eq!(cline["tool_name"], "Bash");
        assert_eq!(cline["session_id"], "conv_1");

        // An agent already speaking the common shape is NOT rewritten, and
        // carries no copy of itself.
        assert_eq!(claude, native, "a native payload travels untouched");
        assert!(claude.get("raw_agent_payload").is_none());
        assert!(codex.get("raw_agent_payload").is_none());

        // The one that WAS remapped carries the original, verbatim.
        assert_eq!(
            cline["raw_agent_payload"]["preToolUse"]["toolName"], "execute_command",
            "the agent's own bytes survive the mapping"
        );

        // What only one of them reports, reported only for that one.
        assert_eq!(codex["turn_id"], "019a5f41");
        assert!(claude.get("turn_id").is_none());
        assert_eq!(codex["model"], "gpt-5.4", "a bare slug stays a bare slug");
        assert_eq!(cline["model"]["provider"], "gemini");
        assert_eq!(cline["agent_version"], "4.1.17");
        assert_eq!(claude["tool_use_id"], "toolu_01");
        // Cline's classic host sends no id, so one is derived from what the
        // pre and the post of a call both carry — and marked as ours.
        assert!(
            cline["tool_use_id"]
                .as_str()
                .is_some_and(|id| id.starts_with("ol_")),
            "a derived id pairs the pre with its post: {cline}"
        );

        // Nothing invented: a pre-call event has no outcome and no duration.
        for d in [&claude, &codex, &cline] {
            assert!(d.get("duration_ms").is_none(), "{d}");
            assert!(d.get("tool_response").is_none(), "{d}");
        }
    }

    /// The mapping is egress-only: the command evaluator reads the RAW
    /// payload, so `run_commands` keeps its per-command semantics — one denied
    /// entry decides, however the list is joined for the wire.
    #[test]
    fn the_command_evaluator_is_unchanged_by_the_common_mapping() {
        let bundle = resident(
            vec![rule("OL-CMD-ENF", "rm -rf*", PolicyRuleMode::Enforce)],
            true,
        );
        let raw = serde_json::json!({
            "taskId": "conv_1",
            "preToolUse": {
                "toolName": "run_commands",
                "parameters": {"commands": "[\"ls\",\"rm -rf /tmp/x\"]"}
            }
        });
        // Anchored rule, second command: only a per-command evaluation matches.
        let decided = crate::core::policy::evaluate(&bundle, "pre_tool_use", Some(&raw))
            .expect("the denied entry decides on the raw payload");
        assert_eq!(decided.rule_id, "OL-CMD-ENF");
        assert!(!decided.shadow);

        // The wire carries the common shape, which is not what was evaluated.
        let common = common_of(raw.clone(), HookEventType::PreToolUse);
        assert_eq!(common["tool_name"], "Bash");
        assert_eq!(common["tool_input"]["command"], "ls\nrm -rf /tmp/x");
        assert_eq!(common["raw_agent_payload"], raw);
    }

    /// A Cline write now reaches the effect table as the `Write` it is —
    /// before the mapping `write_to_file` had no row and produced nothing.
    #[test]
    fn a_cline_write_and_mcp_call_produce_effects() {
        let bundle = resident(vec![], true);
        let envelope = |tool: &str, params: serde_json::Value| -> EventEnvelope {
            serde_json::from_value(serde_json::json!({
                "specversion": "1.0",
                "id": "evt-cline-effect",
                "source": "cline",
                "type": "pre_tool_use",
                "time": "2026-09-15T12:00:00Z",
                "subject": "session-cline",
                "data": {
                    "taskId": "session-cline",
                    "workspaceRoots": ["/work"],
                    "preToolUse": {"toolName": tool, "parameters": params}
                }
            }))
            .expect("valid envelope")
        };
        let facts = crate::zone_eval::facts::FactSet::default();

        let write = map_zone_event(
            &envelope(
                "write_to_file",
                serde_json::json!({"path": "/work/a.rs", "content": "x"}),
            ),
            &bundle,
            None,
        )
        .expect("mapped");
        assert_eq!(write.tool_name, "Write");
        let classified = crate::zone_eval::effect::classify(&write, &facts, None);
        assert!(
            classified.effects.iter().any(|e| e.verb.0 == "write"),
            "a Cline write is a write: {classified:?}"
        );

        let mcp = map_zone_event(
            &envelope(
                "use_mcp_tool",
                serde_json::json!({"server_name": "github", "tool_name": "create_issue", "arguments": "{\"title\":\"x\"}"}),
            ),
            &bundle,
            None,
        )
        .expect("mapped");
        assert_eq!(mcp.tool_name, "mcp__github__create_issue");
        assert_eq!(mcp.tool_input, serde_json::json!({"title": "x"}));
        // An unclassified MCP tool is the Z19 gap it always was for Claude
        // Code — reached now, where before it was an unknown built-in.
        let classified = crate::zone_eval::effect::classify(&mcp, &facts, None);
        assert!(!classified.unknown.is_empty(), "{classified:?}");
    }

    /// A capture-only lane never consumes a queued config alert: its answer is
    /// discarded unread, so the alert would be marked delivered and shown to no
    /// one. It stays queued for the next lane that renders context.
    #[tokio::test]
    async fn a_capture_only_lane_leaves_the_alert_queued() {
        let state = crate::daemon::admin::tests::test_app_state(None, None);
        state
            .pending_alerts
            .push(crate::daemon::config_monitor::PendingAlert {
                alert_id: "alert_1".into(),
                session_ref_id: "sess_c".into(),
                severity: "high".into(),
                headline: "MCP server added".into(),
                body: "Review it before running.".into(),
                source_id: "cline:mcp:abc".into(),
                created_at: "2026-09-29T00:00:00Z".into(),
                delivered: false,
            })
            .await;
        let body = |id: &str| {
            Bytes::from(
                serde_json::json!({
                    "specversion": "1.0",
                    "id": id,
                    "source": "cline",
                    "type": "pre_tool_use",
                    "time": "2026-09-29T00:00:00Z",
                    "subject": "sess_c",
                    "data": {"taskId": "sess_c", "preToolUse": {"toolName": "read_file", "parameters": {"path": id}}}
                })
                .to_string(),
            )
        };
        let headers = |capture_only: bool| {
            let mut h = HeaderMap::new();
            h.insert(CONTENT_TYPE, CT_CLOUDEVENTS_SINGLE.parse().unwrap());
            if capture_only {
                h.insert(
                    crate::hook_output::CAPTURE_ONLY_HEADER,
                    "1".parse().unwrap(),
                );
            }
            h
        };

        let (_, _, Json(captured)) =
            ingest_cloudevent(State(state.clone()), headers(true), body("a")).await;
        assert!(
            captured.context.is_none(),
            "a capture-only lane got the alert"
        );

        let (_, _, Json(rendered)) =
            ingest_cloudevent(State(state.clone()), headers(false), body("b")).await;
        let context = rendered
            .context
            .expect("the alert survived for a rendering lane");
        assert_eq!(context.headline.to_string(), "MCP server added");
    }

    /// Cline's task lifecycle opens sessions too, and is enriched like one.
    #[test]
    fn clines_task_start_and_resume_are_session_starts() {
        for event in [
            HookEventType::SessionStart,
            HookEventType::TaskStart,
            HookEventType::TaskResume,
        ] {
            assert!(starts_a_session(&event), "{event:?}");
        }
        for event in [
            HookEventType::PreToolUse,
            HookEventType::TaskComplete,
            HookEventType::SessionEnd,
        ] {
            assert!(!starts_a_session(&event), "{event:?}");
        }
    }

    /// One tool call reported by both Cline lanes counts ONCE: both reports
    /// are evaluated, only the one that claims the agent's call id commits
    /// session state. Otherwise a "max N writes per session" rule trips at N/2.
    #[tokio::test(start_paused = true)]
    async fn one_call_through_both_lanes_advances_the_counter_once() {
        let mut bundle = resident(vec![], true);
        let zone = crate::zone_eval::load(serde_json::json!({
            "schema_version": 2,
            "organization_id": "org",
            "revision": 1,
            "built_at": "2026-09-10T00:00:00Z",
            "enforcement_enabled": true,
            "signature": null,
            "artifacts": [{
                "artifact_id": "write-cap",
                "atom_id": "atom-write-cap",
                "policy_id": "policy",
                "policy_public_id": "POL-1",
                "dimension": "safety",
                "mode": "enforce",
                "tier": 2,
                "kind": "t2_register_program",
                "on_inconclusive": "ask",
                "hold_mode": "none",
                "spec_hash": "test",
                "body": {
                    "pre": [
                        ["MATCH", {"op": "leaf", "leaf": {
                            "pred": "equals", "field": "effect.verb", "value": "write"
                        }}, 0],
                        ["INC_SAT", 0, 0]
                    ],
                    "state_layout": {"c": 1, "f": 0, "t": 0, "a": 0, "run": false},
                    "verdict": "block",
                    "reason": "count writes",
                    "on_evict": "reinit"
                }
            }]
        }))
        .expect("zone bundle");
        let layout = zone.state_layout;
        bundle.zone = Some(zone);

        let mut app =
            std::sync::Arc::into_inner(crate::daemon::admin::tests::test_app_state(None, None))
                .expect("sole owner");
        app.policy = Some(runtime(Some(bundle), 0));
        let state = Arc::new(app);

        let post = |data: serde_json::Value, id: &str, capture_only: bool| {
            let body = Bytes::from(
                serde_json::json!({
                    "specversion": "1.0", "id": id, "source": "cline",
                    "type": "pre_tool_use", "time": "2026-09-29T00:00:00Z",
                    "subject": "sess_w", "data": data
                })
                .to_string(),
            );
            let mut headers = HeaderMap::new();
            headers.insert(CONTENT_TYPE, CT_CLOUDEVENTS_SINGLE.parse().unwrap());
            if capture_only {
                headers.insert(
                    crate::hook_output::CAPTURE_ONLY_HEADER,
                    "1".parse().unwrap(),
                );
            }
            ingest_cloudevent(State(state.clone()), headers, body)
        };
        let counter = || {
            state
                .policy
                .as_ref()
                .unwrap()
                .sessions
                .preview("sess_w", &layout, std::time::Instant::now())
                .map(|s| s.c)
        };
        let params = serde_json::json!({"path": "/work/a.rs", "content": "x"});

        // The plugin lane, then the file lane, for ONE call.
        let _ = post(
            serde_json::json!({"taskId": "sess_w", "workspaceRoots": ["/work"],
                "toolName": "write_to_file", "toolCallId": "call_1", "parameters": params}),
            "evt-plugin",
            false,
        )
        .await;
        assert_eq!(counter(), Some(vec![1]), "the first report counts");
        let _ = post(
            serde_json::json!({"taskId": "sess_w", "workspaceRoots": ["/work"],
                "hookName": "tool_call",
                "tool_call": {"id": "call_1", "name": "write_to_file", "input": params},
                "preToolUse": {"toolName": "write_to_file", "parameters": params}}),
            "evt-file",
            true,
        )
        .await;
        // The shim's copy waits out the grace, then folds into the plugin's.
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        assert_eq!(
            counter(),
            Some(vec![1]),
            "the second report of one call does not"
        );

        // A different call counts; with no agent id, every report counts as before.
        let _ = post(
            serde_json::json!({"taskId": "sess_w", "workspaceRoots": ["/work"],
                "toolName": "write_to_file", "toolCallId": "call_2", "parameters": params}),
            "evt-2",
            false,
        )
        .await;
        assert_eq!(counter(), Some(vec![2]));
        let _ = post(
            serde_json::json!({"taskId": "sess_w", "workspaceRoots": ["/work"],
                "preToolUse": {"toolName": "write_to_file", "parameters": {"path": "/work/b.rs", "content": "y"}}}),
            "evt-3",
            true,
        )
        .await;
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        assert_eq!(counter(), Some(vec![3]));
    }

    /// A Cline `task_start` through the real pipeline carries the three
    /// `declared_*` keys at the TOP level of `data`, where every consumer reads
    /// them — not inside `raw_agent_payload`, where the mapping used to put
    /// the enrichment it had just been handed.
    #[tokio::test]
    async fn a_cline_task_start_carries_declared_sources_top_level() {
        let mut app =
            std::sync::Arc::into_inner(crate::daemon::admin::tests::test_app_state(None, None))
                .expect("sole owner");
        let (logger, mut logged) = crate::logging::EventLogger::channel();
        app.event_logger = logger;
        let state = Arc::new(app);

        let body = Bytes::from(
            serde_json::json!({
                "specversion": "1.0", "id": "evt-task-start", "source": "cline",
                "type": "task_start", "time": "2026-09-29T00:00:00Z", "subject": "conv_1",
                "data": {
                    "taskId": "conv_1",
                    "hookName": "TaskStart",
                    "clineVersion": "3.0.65",
                    "workspaceRoots": ["/__nonexistent_openlatch_test__/work"],
                    "taskStart": {"taskMetadata": {}}
                }
            })
            .to_string(),
        );
        let mut headers = HeaderMap::new();
        headers.insert(CONTENT_TYPE, CT_CLOUDEVENTS_SINGLE.parse().unwrap());
        let _ = ingest_cloudevent(State(state.clone()), headers, body).await;

        let line = logged.try_recv().expect("the task start was logged");
        let event: serde_json::Value = serde_json::from_str(&line).expect("json");
        let data = &event["data"];
        for key in [
            "openlatch.declared_skills",
            "openlatch.declared_mcp_sources",
            "openlatch.declared_rules_hash",
        ] {
            assert!(data.get(key).is_some(), "{key} missing top-level: {data}");
            assert!(
                data["raw_agent_payload"].get(key).is_none(),
                "{key} nested in the agent's original: {data}"
            );
        }
        assert_eq!(event["type"], "task_start", "the event keeps its own name");
    }

    /// Where Cline loads its plugin, the file shim usually reports a call
    /// first. The stored event must still be the PLUGIN's — the lane that
    /// delivered the block — and where no plugin reports, the shim's is stored
    /// after the grace, recorded as the undelivered verdict it is.
    #[tokio::test(start_paused = true)]
    async fn the_enforcing_lane_is_the_one_stored() {
        let mut bundle = resident(vec![], true);
        bundle.zone = Some(
            crate::zone_eval::load(serde_json::json!({
                "schema_version": 2,
                "organization_id": "org",
                "revision": 1,
                "built_at": "2026-09-10T00:00:00Z",
                "enforcement_enabled": true,
                "signature": null,
                "artifacts": [{
                    "artifact_id": "no-reads",
                    "atom_id": "atom-no-reads",
                    "policy_id": "policy",
                    "policy_public_id": "POL-1",
                    "mode": "enforce",
                    "tier": 1,
                    "kind": "t1_predicate_tree",
                    "hold_mode": "none",
                    "spec_hash": "test",
                    "body": {
                        "node": {"op": "leaf", "leaf": {"pred": "equals", "field": "tool.name", "value": "Read"}},
                        "verdict": "block",
                        "reason": "no reads"
                    }
                }]
            }))
            .expect("zone bundle"),
        );
        let mut app =
            std::sync::Arc::into_inner(crate::daemon::admin::tests::test_app_state(None, None))
                .expect("sole owner");
        let (logger, mut logged) = crate::logging::EventLogger::channel();
        app.event_logger = logger;
        app.policy = Some(runtime(Some(bundle), 0));
        let state = Arc::new(app);

        let post = |data: serde_json::Value, id: &str, capture_only: bool| {
            let body = Bytes::from(
                serde_json::json!({
                    "specversion": "1.0", "id": id, "source": "cline",
                    "type": "pre_tool_use", "time": "2026-09-29T00:00:00Z",
                    "subject": "sess_r", "data": data
                })
                .to_string(),
            );
            let mut headers = HeaderMap::new();
            headers.insert(CONTENT_TYPE, CT_CLOUDEVENTS_SINGLE.parse().unwrap());
            if capture_only {
                headers.insert(
                    crate::hook_output::CAPTURE_ONLY_HEADER,
                    "1".parse().unwrap(),
                );
            }
            ingest_cloudevent(State(state.clone()), headers, body)
        };
        // A distinct path per call: the 100 ms content dedup runs on the real
        // clock, which a paused test does not advance.
        let shim = |call: &str| {
            let params = serde_json::json!({"path": format!("/work/{call}.rs")});
            serde_json::json!({"taskId": "sess_r", "hookName": "tool_call",
                "tool_call": {"id": call, "name": "read_file", "input": params},
                "preToolUse": {"toolName": "read_file", "parameters": params}})
        };
        let stored = |logged: &mut tokio::sync::mpsc::Receiver<String>| {
            let mut out = Vec::new();
            while let Ok(line) = logged.try_recv() {
                out.push(serde_json::from_str::<serde_json::Value>(&line).expect("json"));
            }
            out
        };

        // Shim first, plugin second, one call. The shim is ANSWERED at once —
        // Cline's file hooks are synchronous, so any wait here would be added
        // to every tool call on every host. The paused clock would advance
        // through any sleep on the request path, so zero elapsed proves none.
        let asked = tokio::time::Instant::now();
        let (status, _, Json(shim_answer)) = post(shim("call_9"), "evt-shim", true).await;
        assert_eq!(
            asked.elapsed(),
            Duration::ZERO,
            "the shim's answer never waits"
        );
        assert_eq!(status, StatusCode::OK);
        assert_eq!(shim_answer.verdict.to_string(), "allow");
        assert!(
            stored(&mut logged).is_empty(),
            "only the STORE decision waits"
        );
        let (_, _, Json(answer)) = post(
            serde_json::json!({"taskId": "sess_r", "toolName": "read_file",
                "toolCallId": "call_9", "parameters": {"path": "/work/call_9.rs"}}),
            "evt-plugin",
            false,
        )
        .await;
        assert_eq!(answer.verdict.to_string(), "block");
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        let events = stored(&mut logged);
        assert_eq!(events.len(), 1, "one stored event per call: {events:?}");
        assert_eq!(events[0]["olbinding"], "cline_plugin");
        assert_eq!(events[0]["olresult"], "blocked");
        assert_eq!(events[0]["olenforced"], 1);

        // No plugin: the shim alone is stored after the grace, as undelivered.
        let asked = tokio::time::Instant::now();
        let _ = post(shim("call_10"), "evt-shim-only", true).await;
        assert_eq!(
            asked.elapsed(),
            Duration::ZERO,
            "the shim's answer never waits"
        );
        assert!(
            stored(&mut logged).is_empty(),
            "not stored before the grace"
        );
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        let events = stored(&mut logged);
        assert_eq!(events.len(), 1, "{events:?}");
        assert_eq!(events[0]["olbinding"], "cline_hook");
        assert_eq!(events[0]["olresult"], "allowed");
        assert_eq!(events[0]["olenforced"], 0);
    }

    // -----------------------------------------------------------------------
    // Cline host attribution and the plugin-lane claims
    // -----------------------------------------------------------------------

    /// A state whose resident bundle blocks every `Read`, with the event logger captured — the
    /// fixture of `the_enforcing_lane_is_the_one_stored`, shared by the plugin-lane dedup tests.
    fn no_reads_state() -> (Arc<AppState>, tokio::sync::mpsc::Receiver<String>) {
        let mut bundle = resident(vec![], true);
        bundle.zone = Some(
            crate::zone_eval::load(serde_json::json!({
                "schema_version": 2,
                "organization_id": "org",
                "revision": 1,
                "built_at": "2026-09-10T00:00:00Z",
                "enforcement_enabled": true,
                "signature": null,
                "artifacts": [{
                    "artifact_id": "no-reads",
                    "atom_id": "atom-no-reads",
                    "policy_id": "policy",
                    "policy_public_id": "POL-1",
                    "mode": "enforce",
                    "tier": 1,
                    "kind": "t1_predicate_tree",
                    "hold_mode": "none",
                    "spec_hash": "test",
                    "body": {
                        "node": {"op": "leaf", "leaf": {"pred": "equals", "field": "tool.name", "value": "Read"}},
                        "verdict": "block",
                        "reason": "no reads"
                    }
                }]
            }))
            .expect("zone bundle"),
        );
        let mut app =
            std::sync::Arc::into_inner(crate::daemon::admin::tests::test_app_state(None, None))
                .expect("sole owner");
        let (logger, logged) = crate::logging::EventLogger::channel();
        app.event_logger = logger;
        app.policy = Some(runtime(Some(bundle), 0));
        (Arc::new(app), logged)
    }

    /// One hook POST: `capture_only` sets the file-lane header, `host_pid` the plugin's.
    async fn post_hook(
        state: &Arc<AppState>,
        source: &str,
        type_: &str,
        subject: &str,
        data: serde_json::Value,
        capture_only: bool,
        host_pid: Option<u32>,
    ) -> VerdictResponse {
        let body = Bytes::from(
            serde_json::json!({
                "specversion": "1.0", "id": new_event_id(), "source": source,
                "type": type_, "time": "2026-09-30T00:00:00Z",
                "subject": subject, "data": data
            })
            .to_string(),
        );
        let mut headers = HeaderMap::new();
        headers.insert(CONTENT_TYPE, CT_CLOUDEVENTS_SINGLE.parse().unwrap());
        if capture_only {
            headers.insert(
                crate::hook_output::CAPTURE_ONLY_HEADER,
                "1".parse().unwrap(),
            );
        }
        if let Some(pid) = host_pid {
            headers.insert(
                crate::hook_output::PLUGIN_HOST_PID_HEADER,
                pid.to_string().parse().unwrap(),
            );
        }
        let (_, _, Json(answer)) = ingest_cloudevent(State(state.clone()), headers, body).await;
        answer
    }

    /// Every event the logger has stored so far.
    fn stored_events(logged: &mut tokio::sync::mpsc::Receiver<String>) -> Vec<serde_json::Value> {
        let mut out = Vec::new();
        while let Ok(line) = logged.try_recv() {
            out.push(serde_json::from_str::<serde_json::Value>(&line).expect("json"));
        }
        out
    }

    /// VS Code's file lane: Cline's `PreToolUse` hook payload, which carries no call id.
    fn file_lane_read(subject: &str, path: &str) -> serde_json::Value {
        serde_json::json!({"taskId": subject, "hookName": "PreToolUse",
            "preToolUse": {"toolName": "read_file", "parameters": {"path": path}}})
    }

    /// The plugin lane's report of the same read, with or without its call id.
    fn plugin_read(subject: &str, path: &str, call_id: Option<&str>) -> serde_json::Value {
        let mut data = serde_json::json!({"taskId": subject, "toolName": "read_file",
            "parameters": {"path": path}});
        if let Some(id) = call_id {
            data["toolCallId"] = serde_json::json!(id);
        }
        data
    }

    /// The host is recorded AFTER the verdict. The process-table read is made to block;
    /// the plugin's answer comes back while it is still blocked and before anything is written,
    /// and the seen-state lands once the read is released.
    ///
    /// A plain `#[test]` driving its own runtime: the env locks stay held for the whole test,
    /// and holding a `std::sync::Mutex` guard across an `.await` in an `async fn` is what
    /// `clippy::await_holding_lock` rejects.
    #[test]
    fn seen_is_recorded_after_the_verdict() {
        use crate::hooks::cline_app::process::{
            install_for_tests, test_support::FakeTable, ProcInfo, ProcessTable, Signal,
        };
        use std::sync::atomic::{AtomicBool, Ordering};

        /// Delegates to a FakeTable, but `inspect` parks until the test releases it.
        struct BlockingTable {
            inner: FakeTable,
            entered: AtomicBool,
            release: std::sync::Mutex<std::sync::mpsc::Receiver<()>>,
        }
        impl ProcessTable for BlockingTable {
            fn listener(&self, port: u16) -> Result<Option<u32>, String> {
                self.inner.listener(port)
            }
            fn clients(&self, port: u16) -> Result<Vec<u32>, String> {
                self.inner.clients(port)
            }
            fn inspect(&self, pid: u32) -> Option<ProcInfo> {
                self.entered.store(true, Ordering::SeqCst);
                let _ = self
                    .release
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .recv_timeout(Duration::from_secs(20));
                self.inner.inspect(pid)
            }
            fn spawn_detached(
                &self,
                exe: &std::path::Path,
                args: &[String],
                cwd: &std::path::Path,
                env: &[(String, String)],
            ) -> Result<u32, String> {
                self.inner.spawn_detached(exe, args, cwd, env)
            }
            fn signal(&self, pid: u32, sig: Signal) -> Result<(), String> {
                self.inner.signal(pid, sig)
            }
            fn is_alive(&self, pid: u32) -> bool {
                self.inner.is_alive(pid)
            }
        }

        // The crate's lock order: OPENLATCH_DIR, HOME, the Cline seams, then the process table.
        let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let _home_lock = crate::hooks::claude_code::CONFIG_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let root = tempfile::tempdir().expect("tempdir");
        let ol_dir = root.path().join("openlatch");
        let _seams = crate::hooks::cline::cline_isolated([
            (
                crate::hooks::cline::STORE_DIR_ENV,
                Some(root.path().join("store").into_os_string()),
            ),
            (
                crate::hooks::cline::DATA_DIR_ENV,
                Some(root.path().join("data").into_os_string()),
            ),
            (
                crate::hooks::cline::ASSETS_DIR_ENV,
                Some(root.path().join("assets").into_os_string()),
            ),
            ("HOME", Some(root.path().join("home").into_os_string())),
            ("OPENLATCH_DIR", Some(ol_dir.clone().into_os_string())),
        ]);
        let (release, gate) = std::sync::mpsc::channel();
        let table = Arc::new(BlockingTable {
            inner: FakeTable::new(1),
            entered: AtomicBool::new(false),
            release: std::sync::Mutex::new(gate),
        });
        table.inner.procs.lock().unwrap().insert(
            4242,
            ProcInfo {
                exe: Some(std::path::PathBuf::from(
                    "/Applications/Visual Studio Code.app/Contents/Frameworks/Code Helper (Plugin).app/Contents/MacOS/Code Helper (Plugin)",
                )),
                args: Vec::new(),
                env: None,
                start: None,
                ppid: Some(1),
            },
        );
        let _table = install_for_tests(table.clone());

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .expect("a runtime");
        runtime.block_on(async move {
            let state = crate::daemon::admin::tests::test_app_state(None, None);
            let answer = post_hook(
                &state,
                "cline",
                "pre_tool_use",
                "conv_seen",
                plugin_read("conv_seen", "/work/a.rs", Some("call_1")),
                false,
                Some(4242),
            )
            .await;
            assert_eq!(
                answer.verdict.to_string(),
                "allow",
                "the verdict is answered"
            );

            let seen = crate::daemon::cline_plugin_seen::seen_path(&ol_dir);
            for _ in 0..200 {
                if table.entered.load(Ordering::SeqCst) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            assert!(
                table.entered.load(Ordering::SeqCst),
                "the host is inspected after the verdict"
            );
            assert!(
                !seen.exists(),
                "nothing is recorded while the process-table read is in flight — and the answer \
                 above did not wait for it"
            );

            release.send(()).expect("release the read");
            let mut recorded = None;
            for _ in 0..500 {
                if let Ok(raw) = std::fs::read_to_string(&seen) {
                    recorded = Some(raw);
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            let recorded: serde_json::Value =
                serde_json::from_str(&recorded.expect("the seen-state was written")).expect("json");
            assert!(recorded["vscode"].is_i64(), "{recorded}");
            assert!(recorded.get("hub").is_none() && recorded.get("cli").is_none());
        });
    }

    /// Once the plugin loads on VS Code, a turn's prompt arrives twice — Cline's own
    /// file hook and the plugin's `beforeRun` post, both capture-only. The plugin's (carrying the
    /// host-pid header) is stored at once and claims its prompt; the file lane's waits the grace
    /// and folds into it. With no plugin post, the file lane's prompt is stored after the grace.
    #[tokio::test(start_paused = true)]
    async fn vscode_prompt_from_both_lanes_is_stored_once() {
        let (state, mut logged) = no_reads_state();
        let prompts = |events: &[serde_json::Value]| {
            events
                .iter()
                .filter(|e| e["type"] == "user_prompt_submit")
                .count()
        };

        // The file lane first (Cline's own hook), then the plugin's post.
        let _ = post_hook(
            &state,
            "cline",
            "user_prompt_submit",
            "conv_p",
            serde_json::json!({"taskId": "conv_p", "hookName": "UserPromptSubmit",
                "userPromptSubmit": {"prompt": "list files", "attachments": []}}),
            true,
            None,
        )
        .await;
        assert!(stored_events(&mut logged).is_empty(), "the file lane waits");
        let _ = post_hook(
            &state,
            "cline",
            "user_prompt_submit",
            "conv_p",
            serde_json::json!({"taskId": "conv_p", "hookName": "prompt_submit",
                "userPromptSubmit": {"prompt": "list files", "attachments": []}}),
            true,
            Some(4242),
        )
        .await;
        assert_eq!(
            prompts(&stored_events(&mut logged)),
            1,
            "the plugin's post is stored at once"
        );
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        assert_eq!(
            prompts(&stored_events(&mut logged)),
            0,
            "the file lane's copy folded into the plugin's"
        );

        // No plugin post: the file lane's prompt is the record, after the grace.
        let _ = post_hook(
            &state,
            "cline",
            "user_prompt_submit",
            "conv_p",
            serde_json::json!({"taskId": "conv_p", "hookName": "UserPromptSubmit",
                "userPromptSubmit": {"prompt": "now run the tests", "attachments": []}}),
            true,
            None,
        )
        .await;
        assert!(
            stored_events(&mut logged).is_empty(),
            "not before the grace"
        );
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        assert_eq!(prompts(&stored_events(&mut logged)), 1);
    }

    /// VS Code's file-lane `PreToolUse` carries no call id, and usually arrives
    /// first. It now waits the grace; the plugin's deny is stored, and the file lane's allow
    /// consumes the plugin's content claim and folds into it — whether the plugin's report
    /// carries a call id or not.
    #[tokio::test(start_paused = true)]
    async fn vscode_file_lane_without_id_folds_into_plugin() {
        let (state, mut logged) = no_reads_state();

        for (path, plugin_id) in [("/work/one.rs", Some("call_1")), ("/work/two.rs", None)] {
            let asked = tokio::time::Instant::now();
            let file_answer = post_hook(
                &state,
                "cline",
                "pre_tool_use",
                "conv_v",
                file_lane_read("conv_v", path),
                true,
                None,
            )
            .await;
            assert_eq!(
                asked.elapsed(),
                Duration::ZERO,
                "the file lane is answered at once"
            );
            assert_eq!(file_answer.verdict.to_string(), "allow");
            assert!(
                stored_events(&mut logged).is_empty(),
                "the id-less file lane is no longer stored at once"
            );

            let plugin_answer = post_hook(
                &state,
                "cline",
                "pre_tool_use",
                "conv_v",
                plugin_read("conv_v", path, plugin_id),
                false,
                None,
            )
            .await;
            assert_eq!(plugin_answer.verdict.to_string(), "block");

            tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
            let events = stored_events(&mut logged);
            assert_eq!(events.len(), 1, "one stored event per call: {events:?}");
            assert_eq!(events[0]["olbinding"], "cline_plugin");
            assert_eq!(events[0]["olresult"], "blocked");
            assert!(
                !state.plugin_claims.consume(
                    "conv_v",
                    &dedup_identity(&plugin_read("conv_v", path, plugin_id))
                ),
                "the file lane consumed the plugin's claim"
            );
        }
    }

    /// The miss: with no plugin claim the file-lane event is the record, stored after
    /// the grace as the undelivered verdict it is. And a header-less file-lane prompt is still
    /// OBSERVED at arrival: its run is covered, so no session-store backfill is scheduled.
    #[tokio::test(start_paused = true)]
    async fn file_lane_without_plugin_is_still_stored() {
        let (state, mut logged) = no_reads_state();

        let _ = post_hook(
            &state,
            "cline",
            "pre_tool_use",
            "conv_f",
            file_lane_read("conv_f", "/work/alone.rs"),
            true,
            None,
        )
        .await;
        assert!(
            stored_events(&mut logged).is_empty(),
            "not before the grace"
        );
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        let events = stored_events(&mut logged);
        assert_eq!(events.len(), 1, "{events:?}");
        assert_eq!(events[0]["olbinding"], "cline_hook");
        assert_eq!(events[0]["olresult"], "allowed");

        // The backfill half. The task end is observed directly — posting it would spawn a real
        // session-store read for a job that must not exist.
        let task_complete = |subject: &str| -> EventEnvelope {
            serde_json::from_value(serde_json::json!({
                "specversion": "1.0", "id": new_event_id(), "source": "cline",
                "type": "task_complete", "time": "2026-09-30T00:00:00Z", "subject": subject,
                "data": {"taskId": subject, "sessionContext": {"rootSessionId": "session_f"}}
            }))
            .expect("an envelope")
        };
        for (subject, with_prompt) in [("conv_b", true), ("conv_c", false)] {
            let _ = post_hook(
                &state,
                "cline",
                "task_start",
                subject,
                serde_json::json!({"taskId": subject, "hookName": "TaskStart",
                    "sessionContext": {"rootSessionId": "session_f"}}),
                true,
                None,
            )
            .await;
            if with_prompt {
                let _ = post_hook(
                    &state,
                    "cline",
                    "user_prompt_submit",
                    subject,
                    serde_json::json!({"taskId": subject, "hookName": "UserPromptSubmit",
                        "userPromptSubmit": {"prompt": "hi", "attachments": []}}),
                    true,
                    None,
                )
                .await;
            }
            let job = state.cline_prompts.observe(
                &task_complete(subject),
                chrono::Utc::now().timestamp_millis(),
            );
            assert_eq!(
                job.is_none(),
                with_prompt,
                "{subject}: a prompt observed at arrival covers its run; none leaves it to the \
                 backfill: {job:?}"
            );
        }
    }

    /// A claim is a COUNT. Call 1 through both lanes folds one-for-one; call 2 — the same
    /// tool and input, inside the claim TTL — reaches only the file lane, finds no claim left and
    /// is stored. Two stored events for two calls.
    #[tokio::test(start_paused = true)]
    async fn second_identical_call_without_plugin_survives() {
        let (state, mut logged) = no_reads_state();
        let path = "/work/same.rs";

        let _ = post_hook(
            &state,
            "cline",
            "pre_tool_use",
            "conv_s",
            file_lane_read("conv_s", path),
            true,
            None,
        )
        .await;
        let _ = post_hook(
            &state,
            "cline",
            "pre_tool_use",
            "conv_s",
            plugin_read("conv_s", path, Some("call_1")),
            false,
            None,
        )
        .await;
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        let first = stored_events(&mut logged);
        assert_eq!(first.len(), 1, "{first:?}");
        assert_eq!(first[0]["olbinding"], "cline_plugin");

        // Past the 100 ms content dedup, which runs on the real clock a paused test never
        // advances — well inside the 10 s claim TTL.
        std::thread::sleep(Duration::from_millis(150));
        let _ = post_hook(
            &state,
            "cline",
            "pre_tool_use",
            "conv_s",
            file_lane_read("conv_s", path),
            true,
            None,
        )
        .await;
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        let second = stored_events(&mut logged);
        assert_eq!(
            second.len(),
            1,
            "the repeated call keeps its file event: {second:?}"
        );
        assert_eq!(second[0]["olbinding"], "cline_hook");
        assert_eq!(first.len() + second.len(), 2);
    }

    /// Deferring is Cline's alone: a Claude Code or Codex capture-only report with no call id is
    /// processed at once, exactly as before — nothing deferred, nothing claimed.
    #[tokio::test(start_paused = true)]
    async fn claude_and_codex_capture_only_unchanged() {
        let (state, mut logged) = no_reads_state();
        for (source, subject) in [("claude-code", "sess_cc"), ("codex-cli", "sess_cx")] {
            let asked = tokio::time::Instant::now();
            let _ = post_hook(
                &state,
                source,
                "pre_tool_use",
                subject,
                serde_json::json!({"session_id": subject, "tool_name": "Bash",
                    "tool_input": {"command": format!("ls {source}")}}),
                true,
                None,
            )
            .await;
            let _ = post_hook(
                &state,
                source,
                "user_prompt_submit",
                subject,
                serde_json::json!({"session_id": subject, "prompt": format!("hello {source}")}),
                true,
                None,
            )
            .await;
            assert_eq!(asked.elapsed(), Duration::ZERO);
            let events = stored_events(&mut logged);
            assert_eq!(
                events.len(),
                2,
                "{source}: stored at once, never deferred: {events:?}"
            );
        }
        tokio::time::sleep(CAPTURE_TWIN_GRACE + Duration::from_millis(100)).await;
        assert!(
            stored_events(&mut logged).is_empty(),
            "and nothing was left behind to be stored after a grace"
        );
    }

    /// The post-call half: outcome and duration, however the agent phrases it.
    #[test]
    fn the_outcome_is_read_from_whichever_half_the_agent_reports() {
        let cline = common_of(
            serde_json::json!({
                "taskId": "conv_1",
                "postToolUse": {
                    "toolName": "execute_command",
                    "result": "boom",
                    "success": false,
                    "executionTimeMs": 1234
                }
            }),
            HookEventType::PostToolUse,
        );
        assert_eq!(cline["duration_ms"], 1234);
        assert_eq!(
            cline["tool_response"],
            serde_json::json!({"content": "boom", "is_error": true}),
            "Cline's `success` beside the result becomes the common `is_error`"
        );
        assert_eq!(cline["raw_agent_payload"]["postToolUse"]["success"], false);

        // Claude Code and Codex describe the outcome inside the response, and
        // already speak the common shape — so nothing is rewritten.
        let native = serde_json::json!({"tool_name": "Bash", "tool_response": {"is_error": true}});
        let claude = common_of(native.clone(), HookEventType::PostToolUse);
        assert_eq!(claude, native);
    }

    /// Dedup compares the action, not the bytes — so Cline's two lanes, which
    /// describe one shell command in two shapes, collapse into one event.
    #[test]
    fn both_cline_lanes_dedup_to_one_action() {
        let file_lane = serde_json::json!({
            "taskId": "conv_1",
            "hookName": "PreToolUse",
            "clineVersion": "4.1.17",
            "preToolUse": {"toolName": "execute_command", "parameters": {"command": "rm -rf /tmp"}}
        });
        let plugin_lane = serde_json::json!({
            "toolName": "execute_command",
            "parameters": {"command": "rm -rf /tmp"}
        });
        assert_eq!(
            dedup_identity(&file_lane),
            dedup_identity(&plugin_lane),
            "one command, two lanes, one action"
        );

        // A different command is still a different event.
        let other =
            serde_json::json!({"toolName": "execute_command", "parameters": {"command": "ls"}});
        assert_ne!(dedup_identity(&plugin_lane), dedup_identity(&other));

        // An event with neither a tool nor a prompt keeps hashing its whole
        // payload: its body IS its identity.
        let start = serde_json::json!({"hookName": "TaskStart", "taskId": "conv_1"});
        assert_eq!(dedup_identity(&start), start);
    }

    /// One prompt, posted by Cline's plugin from `beforeRun` and by the
    /// `UserPromptSubmit` file hook, is one prompt — whatever else each payload
    /// carries.
    #[test]
    fn both_cline_prompt_lanes_dedup_to_one_prompt() {
        let file_lane = serde_json::json!({
            "hookName": "prompt_submit",
            "taskId": "conv_1",
            "clineVersion": "",
            "workspaceRoots": ["/w"],
            "timestamp": "2026-09-23T10:00:00.000Z",
            "userPromptSubmit": {"prompt": "list files", "attachments": []}
        });
        let plugin_lane = serde_json::json!({
            "hookName": "prompt_submit",
            "taskId": "conv_1",
            "agent_id": "agent_1",
            "timestamp": "2026-09-23T10:00:00.004Z",
            "userPromptSubmit": {"prompt": "list files", "attachments": []}
        });
        assert_eq!(dedup_identity(&file_lane), dedup_identity(&plugin_lane));

        // Claude Code's flat spelling reads the same way.
        let flat = serde_json::json!({"prompt": "list files", "session_id": "s"});
        assert_eq!(
            dedup_identity(&flat),
            serde_json::json!({"prompt": "list files"})
        );

        // A different prompt is still a different event.
        let other = serde_json::json!({"userPromptSubmit": {"prompt": "run ls"}});
        assert_ne!(dedup_identity(&plugin_lane), dedup_identity(&other));
    }

    #[test]
    fn the_root_session_is_read_by_shape_wherever_the_payload_keeps_it() {
        let top = serde_json::json!({"taskId": "conv_1", "sessionContext": {"rootSessionId": "session_1"}});
        assert_eq!(root_session_id_of(&top), Some("session_1"));
        let nested = serde_json::json!({"raw_agent_payload": {"sessionContext": {"rootSessionId": " session_2 "}}});
        assert_eq!(root_session_id_of(&nested), Some("session_2"));
        for none in [
            serde_json::json!({"session_id": "sess_a"}),
            serde_json::json!({"sessionContext": {"rootSessionId": ""}}),
            serde_json::json!({"sessionContext": {"rootSessionId": 7}}),
        ] {
            assert_eq!(root_session_id_of(&none), None, "{none}");
        }
    }

    /// A registry holding one live session of `source`'s.
    fn registry_with(
        source: &str,
        session_id: &str,
    ) -> crate::model_relay::session::SessionRegistry {
        let registry = crate::model_relay::session::SessionRegistry::default();
        registry.upsert_signals(
            "agt_install",
            "agt_install",
            source,
            session_id,
            &crate::model_relay::session::SessionSignals::default(),
        );
        registry
    }

    /// The agent named its session, so nothing is resolved and nothing is
    /// replaced — the subject it sent is the one that ships.
    #[test]
    fn a_named_session_is_attested_and_left_alone() {
        let registry = registry_with("cline", "conv_live");
        let (subject, assurance) =
            session_stamp(ProcessMode::Live, true, &registry, "agt_install", "cline");
        assert_eq!(subject, None, "a named session is never overwritten");
        assert_eq!(assurance, Some(Assurance::Attested));
    }

    /// **The Cline plugin lane.** It sends `{toolName, parameters}` — no task
    /// id, no session of any kind — so the hook binary wrote
    /// [`SESSIONLESS_SUBJECT`] and every such event in the fleet shared it.
    ///
    /// Correlated to the live session of its OWN agent, and labelled `inferred`
    /// however few candidates there were: nothing named it, so it is a guess.
    #[test]
    fn a_sessionless_event_is_correlated_but_never_attested() {
        let registry = registry_with("cline", "conv_live");
        let (subject, assurance) =
            session_stamp(ProcessMode::Live, false, &registry, "agt_install", "cline");
        assert_eq!(subject.as_deref(), Some("conv_live"));
        assert_eq!(
            assurance,
            Some(Assurance::Inferred),
            "one live session makes the guess a good one, not a fact"
        );
    }

    /// The correlation never crosses agents. A Cline tool call with no session
    /// of its own must not land on a live Claude Code session — that pair is the
    /// platform's `(org, source, agent_id)` join key, and a wrong one is worse
    /// than none.
    #[test]
    fn the_correlation_is_scoped_to_the_events_own_agent() {
        let registry = registry_with("claude-code", "sess_claude");
        let (subject, assurance) =
            session_stamp(ProcessMode::Live, false, &registry, "agt_install", "cline");
        assert_eq!(
            subject.as_deref(),
            Some("agt_install"),
            "no Cline session is live, so the install bucket — never the Claude session"
        );
        assert_eq!(assurance, Some(Assurance::Unknown));
    }

    /// Nothing live at all: the install is a stable bucket the platform can
    /// render as host-level, and `unknown` says plainly that no session was
    /// claimed.
    #[test]
    fn nothing_live_falls_back_to_the_install_bucket() {
        let registry = crate::model_relay::session::SessionRegistry::default();
        let (subject, assurance) =
            session_stamp(ProcessMode::Live, false, &registry, "agt_install", "cline");
        assert_eq!(subject.as_deref(), Some("agt_install"));
        assert_eq!(assurance, Some(Assurance::Unknown));

        // An install with no agent id configured has no bucket to offer, and
        // inventing one would be worse than the placeholder.
        let (subject, assurance) = session_stamp(ProcessMode::Live, false, &registry, "", "cline");
        assert_eq!(subject, None);
        assert_eq!(assurance, Some(Assurance::Unknown));
    }

    /// Replay resolves nothing and stamps nothing: today's live session is not
    /// the session of a hook call that finished hours ago. Same rule as
    /// `olpolicy*` and the proxy attributes.
    #[test]
    fn replay_is_never_session_stamped() {
        let registry = registry_with("cline", "conv_live");
        let (subject, assurance) = session_stamp(
            ProcessMode::Replay,
            false,
            &registry,
            "agt_install",
            "cline",
        );
        assert_eq!(subject, None);
        assert_eq!(assurance, None);
    }

    fn fake_signals() -> IdentitySignals {
        IdentitySignals {
            git_email: Some("dev@example.com".to_string()),
            provider_account: Some("dev@anthropic.example".to_string()),
        }
    }

    fn stamp(
        mode: ProcessMode,
        capture: bool,
        subject: Option<&str>,
    ) -> (Option<String>, IdentitySignals) {
        identity_stamp(
            mode,
            capture,
            subject,
            Some("/repo"),
            "claude-code",
            || Some("injected-os-user".to_string()),
            |_, _, _| fake_signals(),
        )
    }

    #[test]
    fn capture_on_with_a_session_stamps_all_three() {
        let (osuser, signals) = stamp(ProcessMode::Live, true, Some("sess_a"));
        assert_eq!(osuser.as_deref(), Some("injected-os-user"));
        assert_eq!(signals, fake_signals());
    }

    /// D-02's two indistinguishable inputs. `capture == false` is what the
    /// caller folds "no resident bundle" into, so this one test covers a client
    /// that has never fetched a bundle and one whose org left the switch off.
    #[test]
    fn capture_off_or_no_bundle_stamps_nothing() {
        let (osuser, signals) = stamp(ProcessMode::Live, false, Some("sess_a"));
        assert_eq!(osuser, None);
        assert_eq!(signals, IdentitySignals::default());
    }

    /// D-09. Replay is suppressed by the gate itself, not merely by the absence
    /// of a bundle on that path — so this stays true if a future change starts
    /// loading policy for replayed envelopes.
    #[test]
    fn replay_is_never_identity_stamped_even_with_capture_on() {
        let (osuser, signals) = stamp(ProcessMode::Replay, true, Some("sess_a"));
        assert_eq!(osuser, None);
        assert_eq!(signals, IdentitySignals::default());
    }

    /// A sessionless hook still has a host behind it. `gitemail` and
    /// `provideracct` have nothing to key on without a subject, and the session
    /// cache is never touched — asserted by a lookup closure that would panic if
    /// it were.
    #[test]
    fn a_sessionless_envelope_carries_osuser_only() {
        let (osuser, signals) = identity_stamp(
            ProcessMode::Live,
            true,
            None,
            Some("/repo"),
            "claude-code",
            || Some("injected-os-user".to_string()),
            |_, _, _| panic!("no subject means no session lookup"),
        );
        assert_eq!(osuser.as_deref(), Some("injected-os-user"));
        assert_eq!(signals, IdentitySignals::default());
    }

    /// The agent platform reaches the resolver, because it is what decides
    /// whether the provider account — read from Claude Code's own state file —
    /// is looked up for this session at all.
    #[test]
    fn the_events_source_reaches_the_resolver() {
        let (_, signals) = identity_stamp(
            ProcessMode::Live,
            true,
            Some("sess_a"),
            None,
            "cline",
            || None,
            |_, _, source| IdentitySignals {
                git_email: Some(source.to_string()),
                provider_account: None,
            },
        );
        assert_eq!(signals.git_email.as_deref(), Some("cline"));
    }

    /// The working directory reaching the resolver is the one this event
    /// reported — it is the git-email lookup's only input.
    #[test]
    fn the_events_cwd_reaches_the_resolver() {
        let (_, signals) = identity_stamp(
            ProcessMode::Live,
            true,
            Some("sess_a"),
            Some("/repo/sub"),
            "claude-code",
            || None,
            |session_id, cwd, _| {
                assert_eq!(session_id, "sess_a");
                IdentitySignals {
                    git_email: cwd.map(str::to_owned),
                    provider_account: None,
                }
            },
        );
        assert_eq!(signals.git_email.as_deref(), Some("/repo/sub"));
    }

    // -- the cloud status word and the last refusal (I-4) --------------------

    /// Table over the ladder, in order. Never the async handler: these are the
    /// two pure helpers, and the ORDER is what is under test — a word that
    /// jumped the queue would mask the one above it.
    #[test]
    fn cloud_status_word_reports_the_ladder_in_order() {
        use std::sync::atomic::Ordering;

        let state = crate::cloud::CloudState::new();
        state.set_no_credential(false);
        assert_eq!(cloud_status_word(&state), "connected");

        state.record_drop();
        assert_eq!(cloud_status_word(&state), "network_error");

        state.set_emergency_mode(true);
        assert_eq!(cloud_status_word(&state), "emergency");

        state.set_no_credential(true);
        assert_eq!(cloud_status_word(&state), "no_credential");

        state.set_license_gate("license_expired".to_string(), None, Some(Duration::ZERO));
        assert_eq!(
            cloud_status_word(&state),
            "license_blocked",
            "a gate outranks every word below it — and the cadence having \
             elapsed does not make it healthy: the next POST is due, not proven"
        );

        state.auth_error.store(true, Ordering::Relaxed);
        assert_eq!(
            cloud_status_word(&state),
            "auth_error",
            "a rejected credential is the one thing that outranks a refusal"
        );

        // A cleared gate falls back through the ladder rather than latching.
        state.auth_error.store(false, Ordering::Relaxed);
        assert!(state.clear_license_gate());
        assert_eq!(cloud_status_word(&state), "no_credential");
    }

    #[test]
    fn cloud_last_refusal_is_null_until_there_is_one() {
        assert_eq!(cloud_last_refusal_json(None), serde_json::Value::Null);

        let state = crate::cloud::CloudState::new();
        state.set_license_gate(
            "site_license_expired".to_string(),
            Some("https://app.openlatch.ai/settings/licensing".to_string()),
            None,
        );
        let gate = state.license_gate().expect("gate");
        let json = cloud_last_refusal_json(Some(&gate));

        assert_eq!(json["code"], "site_license_expired");
        assert_eq!(
            json["licensing_url"],
            "https://app.openlatch.ai/settings/licensing"
        );
        let at = json["at"].as_str().expect("`at` is a string");
        assert!(
            chrono::DateTime::parse_from_rfc3339(at).is_ok(),
            "`at` must be RFC 3339, got {at}"
        );
    }

    // -- host stamping (I-4) -------------------------------------------------

    const FAKE_HOST_ID: &str = "9262b296baa37a205734509345581251";

    /// Injected closures throughout — never `identity::host_id` /
    /// `identity::hostname`, which memoise process-wide and would freeze
    /// whatever this test saw first for the rest of the binary.
    fn host(mode: ProcessMode, capture: bool) -> (Option<String>, Option<String>) {
        host_stamp(
            mode,
            capture,
            None,
            || Some(FAKE_HOST_ID.to_string()),
            || Some("devbox-01".to_string()),
        )
    }

    #[test]
    fn live_with_capture_stamps_the_host_id_and_the_hostname() {
        let (hostid, hostname) = host(ProcessMode::Live, true);
        assert_eq!(hostid.as_deref(), Some(FAKE_HOST_ID));
        assert_eq!(hostname.as_deref(), Some("devbox-01"));
    }

    /// The asymmetry that is the whole point of this function: `hostid` is a
    /// machine fact and licensing needs it whatever an organization decides
    /// about identity capture, so it survives both gates that suppress
    /// `hostname`.
    #[test]
    fn host_id_survives_capture_off_and_replay_while_hostname_does_not() {
        let (hostid, hostname) = host(ProcessMode::Live, false);
        assert_eq!(hostid.as_deref(), Some(FAKE_HOST_ID));
        assert_eq!(hostname, None);

        let (hostid, hostname) = host(ProcessMode::Replay, true);
        assert_eq!(hostid.as_deref(), Some(FAKE_HOST_ID));
        assert_eq!(hostname, None);
    }

    /// A host with no machine identifier — many containers — sends no
    /// `hostid` at all rather than an empty or invented one.
    #[test]
    fn a_host_with_no_machine_identifier_stamps_no_host_id() {
        let (hostid, hostname) = host_stamp(
            ProcessMode::Live,
            true,
            None,
            || None,
            || Some("devbox-01".to_string()),
        );
        assert_eq!(hostid, None);
        assert_eq!(hostname.as_deref(), Some("devbox-01"));
    }

    /// A producer-supplied value is kept, and the resolver is never reached —
    /// asserted by a closure that would panic if it were.
    #[test]
    fn a_producer_supplied_host_id_is_kept_and_never_re_resolved() {
        let (hostid, _) = host_stamp(
            ProcessMode::Live,
            true,
            Some("deadbeefdeadbeefdeadbeefdeadbeef".to_string()),
            || panic!("an envelope that carries a hostid must not re-resolve one"),
            || None,
        );
        assert_eq!(hostid.as_deref(), Some("deadbeefdeadbeefdeadbeefdeadbeef"));
    }
}