openlatch-client 0.5.4

//! 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::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::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 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 (
            ev_type,
            ev_id,
            ev_subject,
            dedup_hit,
            policy_match,
            zone_decision,
            hold_wait,
            delivery,
        ) = process_envelope(state.clone(), envelope, start, ProcessMode::Live).await;
        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.
    if !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.
#[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).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,
) -> (
    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());
    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(&event.session_id, zone.state_layout, now, |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"
        );
    }

    // 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(),
        envelope.subject.as_deref(),
    ) {
        // 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"),
        };
        state.registry.upsert_signals(
            install_id,
            install_id,
            envelope.source.as_str(),
            session_id,
            &signals,
        );
    }

    // 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.
    let event_cwd = envelope
        .data
        .as_ref()
        .and_then(|d| d.get("cwd"))
        .and_then(|v| v.as_str());

    // 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 (osuser, identity) = identity_stamp(
        mode,
        resident.is_some_and(|b| b.capture_identity_signals),
        envelope.subject.as_deref(),
        event_cwd,
        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.
    let config_monitor_tx = state.config_monitor.request_tx();
    if matches!(event_type, HookEventType::SessionStart) {
        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(cwd) = event_cwd {
        // 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.
        crate::daemon::config_monitor::try_register_project_from_cwd(
            cwd,
            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 = envelope.data.clone().unwrap_or_default();
    let is_duplicate = state
        .dedup
        .check_and_insert(&subject, &type_str, &data_for_dedup);

    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());
    }
    // 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),
        );

        // NORMALIZED AGENT FACTS, read from the payload by SHAPE and stamped
        // here for the same reason the rest of the `ol*` family is: one egress,
        // every envelope, undeclared in the schema.
        //
        // `data` is NOT touched — the raw payload still rides verbatim, which
        // is the envelope's standing contract. These are additive attributes
        // beside it, so a consumer reads one field whatever agent produced the
        // event instead of carrying a parser per agent. Three shipped agents
        // already disagree about where the prompt lives (`prompt` for Claude
        // Code and Codex, `userPromptSubmit.prompt` for Cline), and that
        // knowledge belongs in one place rather than in every consumer.
        //
        // Absent means absent: a payload with no prompt gets no attribute
        // rather than an empty string, so "did this event carry a prompt" is
        // answerable.
        {
            use crate::core::envelope::normalize;
            let data = obj.get("data").cloned().unwrap_or(serde_json::Value::Null);
            if let Some(prompt) = normalize::prompt_of(&data) {
                obj.insert("olprompt".to_string(), serde_json::json!(prompt));
            }
            let (provider, slug) = normalize::model_of(&data);
            if let Some(provider) = provider {
                obj.insert("olmodelprovider".to_string(), serde_json::json!(provider));
            }
            if let Some(slug) = slug {
                obj.insert("olmodelslug".to_string(), serde_json::json!(slug));
            }
        }
        // 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(),
                    &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,
                    ProcessMode::Replay => serde_json::to_value(&envelope).unwrap_or_default(),
                },
                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,
    )
}

/// 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()?;
    // Cline spells these two `parameters` and `toolName`. The fallback is
    // 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. Same `.or_else()` idiom `tool_result`
    // uses immediately below, for the same reason: one extraction, several
    // agents' spellings, no per-agent branch.
    let tool_input = object
        .get("tool_input")
        .cloned()
        .or_else(|| object.get("parameters").cloned())
        .unwrap_or(serde_json::Value::Null);
    let tool_result = object
        .get("tool_result")
        .cloned()
        .or_else(|| object.get("result").cloned())
        .or_else(|| object.get("tool_response").cloned())
        .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 = object
        .get("cwd")
        .and_then(serde_json::Value::as_str)
        .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: object
            .get("tool_use_id")
            .and_then(serde_json::Value::as_str)
            .unwrap_or_default()
            .to_string(),
        tool_name: object
            .get("tool_name")
            .and_then(serde_json::Value::as_str)
            .or_else(|| object.get("toolName").and_then(serde_json::Value::as_str))
            .unwrap_or_default()
            .to_string(),
        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),
    })
}

fn stamp_record_v2(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    decision: &ZoneDecision,
    bundle: Option<&ResidentBundle>,
    source: &str,
    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!(format!("{}_hook", source.replace('-', "_"))),
    );
    let (enforced, result) = delivery
        .and_then(|delivery| delivery.enforced.zip(delivery.result))
        .unwrap_or_else(|| delivery_outcome(source, 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.
fn delivery_outcome(
    source: &str,
    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);
    let supported_agent = matches!(source, "claude-code" | "codex-cli");
    match decision.verdict {
        Verdict::Allow => (true, "allowed"),
        Verdict::Ask => (pre_action && supported_agent, "flagged"),
        Verdict::Block if pre_action && supported_agent => (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,
    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, 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,
) {
    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(
            "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));
}

/// 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`.
///
/// 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>,
    resolve_os_user: impl FnOnce() -> Option<String>,
    observe_session: impl FnOnce(&str, Option<&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),
        None => IdentitySignals::default(),
    };
    (resolve_os_user(), signals)
}

/// 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(),
    }))
}

/// 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();
        let status = if cs.is_auth_error() {
            "auth_error"
        } else if cs.is_no_credential() {
            "no_credential"
        } else if emergency {
            "emergency"
        } else if cs.consecutive_drops() > 0 {
            "network_error"
        } else {
            "connected"
        };
        (
            status,
            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 (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 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),
    };

    Json(serde_json::json!({
        "events_processed": events,
        "uptime_secs": uptime_secs,
        "update_available": update_available,
        "cloud_status": cloud_status,
        "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,
        "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(),
    }))
}

/// `/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,
    /// `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,
    /// 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,
}

/// 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,
            bundle_age_seconds: serde_json::Value::Null,
            last_poll_ok_secs: 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,
        };
    };
    let resident = policy.handle.load_full();
    let (
        has_bundle,
        schema_version,
        bundle_digest,
        revision,
        bundle_age_seconds,
        agent_function,
        omitted_kinds,
        skipped_items,
    ) = 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()),
        ),
        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,
        ),
    };
    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
    };
    PolicyMetrics {
        enabled: true,
        has_bundle,
        schema_version,
        bundle_digest,
        revision,
        bundle_age_seconds,
        last_poll_ok_secs,
        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,
    }
}

/// 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");

        assert_eq!(mapped.tool_name, "run_commands");
        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",
            &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",
            &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",
            &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",
            &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",
                &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",
            &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!(
            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("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 {
        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)),
            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()),
        }
    }

    // --- 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,
        };
        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");

        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);
    }

    /// 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);
    }

    // -- 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(),
            5,
            "five of the six; 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.
    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"),
            || 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"),
            || 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 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"),
            || 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"));
    }
}