openlatch-client 0.1.18

//! 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;

/// 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 {
        let (ev_type, ev_id, ev_subject, dedup_hit, policy_match) =
            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, policy_match.as_ref()),
                event_type: ev_type,
                event_id: ev_id,
                policy_match,
            },
        );
    }

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

    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::Deny => {
            let m = outcome.policy_match.as_ref();
            VerdictResponse::deny(
                outcome.event_id,
                latency_ms,
                m.map(|m| m.reason.clone()),
                m.map(|m| m.severity.to_string()),
                m.map(|m| m.rule_id.clone()),
            )
        }
        Verdict::Approve => VerdictResponse::approve(outcome.event_id, latency_ms),
        Verdict::Allow => VerdictResponse::allow(outcome.event_id, latency_ms),
    };

    // 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::Deny) && !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))
}

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

/// Rank in the most-restrictive-wins lattice: `deny > approve > allow`.
fn verdict_rank(verdict: Verdict) -> u8 {
    match verdict {
        Verdict::Allow => 0,
        Verdict::Approve => 1,
        Verdict::Deny => 2,
    }
}

/// 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 denies. Everything else is exactly what the daemon
/// did before policy existed: `Stop` is approved, all else allowed. A shadow
/// match ("would have denied") never changes the verdict — it only annotates
/// the stored event with `olverdictshadow`.
fn verdict_for(event_type: &HookEventType, policy_match: Option<&PolicyMatch>) -> Verdict {
    if policy_match.is_some_and(|m| !m.shadow) {
        return Verdict::Deny;
    }
    match event_type {
        HookEventType::Stop => Verdict::Approve,
        _ => 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 its existing `allow`/`approve` value — the
/// closed enum has no `ask` variant, 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) =
        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>,
) {
    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())
    });
    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, Some(m)),
            shadow = m.shadow,
            revision = bundle.revision,
            "policy decision"
        );
    }

    // Model-boundary (D-09): stamp the shared session registry so the boundary
    // 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(),
    ) {
        state
            .registry
            .upsert(install_id, install_id, envelope.source.as_str(), session_id);
    }

    // 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 boundary 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.
    if matches!(event_type, HookEventType::SessionStart) {
        if let Err(e) = crate::daemon::config_monitor::enrich_session_start(
            &mut envelope.data,
            &state.content_hash_cache,
            state.config_monitor_request_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,
            state.config_monitor_request_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);
    }

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

    (event_type, event_id, raw_subject, false, policy_match)
}

/// 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` | `"deny"` | 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::Deny.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)
}

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": { "boundary": {"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 boundary 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"
        );
    }
    Json(serde_json::json!({
        "status": if degraded.is_empty() { "ok" } else { "degraded" },
        // 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"),
        "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());

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

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

/// 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,
            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,
        };
    };
    let resident = policy.handle.load_full();
    let (has_bundle, revision, bundle_age_seconds, agent_function) = match resident.as_ref() {
        Some(b) => (
            true,
            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),
        ),
        None => (
            false,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
        ),
    };
    // 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,
        revision,
        bundle_age_seconds,
        last_poll_ok_secs,
        agent_function,
    }
}

/// 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 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::Approve => HookEventType::Stop,
                _ => HookEventType::PreToolUse,
            },
            event_id: id.to_string(),
            policy_match: match verdict {
                Verdict::Deny => Some(match_of("OL-CMD-ENF", false)),
                _ => None,
            },
        }
    }

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

    #[test]
    fn a_non_shadow_match_denies_and_a_shadow_match_does_not() {
        assert_eq!(
            verdict_for(
                &HookEventType::PreToolUse,
                Some(&match_of("OL-CMD-A", false))
            ),
            Verdict::Deny
        );
        assert_eq!(
            verdict_for(
                &HookEventType::PreToolUse,
                Some(&match_of("OL-CMD-A", true))
            ),
            Verdict::Allow,
            "observe never blocks"
        );
        assert_eq!(
            verdict_for(&HookEventType::PreToolUse, None),
            Verdict::Allow
        );
        assert_eq!(verdict_for(&HookEventType::Stop, None), Verdict::Approve);
    }

    /// A deny outranks the Stop→approve upgrade, even on a Stop envelope.
    #[test]
    fn deny_outranks_the_stop_upgrade() {
        assert_eq!(
            verdict_for(&HookEventType::Stop, Some(&match_of("OL-CMD-A", false))),
            Verdict::Deny
        );
    }

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

    #[test]
    fn the_fold_is_most_restrictive_wins() {
        for (a, b) in [
            (Verdict::Allow, Verdict::Approve),
            (Verdict::Allow, Verdict::Deny),
            (Verdict::Approve, Verdict::Deny),
        ] {
            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::Deny, Verdict::Approve, Verdict::Allow],
            [Verdict::Allow, Verdict::Deny, Verdict::Approve],
            [Verdict::Approve, Verdict::Allow, Verdict::Deny],
        ];
        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::Deny, "{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::Deny, "evt_2"));
        let got = worst.expect("folded");
        assert_eq!(got.event_id, "evt_2");
        assert_eq!(
            got.policy_match.expect("deny 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,
        );
        assert_eq!(obj.get("olverdictshadow").unwrap(), "deny");
        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)),
        }
    }

    #[test]
    fn metrics_report_policy_disabled_as_such() {
        let m = policy_metrics(None, SystemTime::now());
        assert!(!m.enabled);
        assert!(!m.has_bundle);
        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.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.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 deny 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 deny.
    #[test]
    fn attaching_an_alert_to_a_deny_would_overwrite_the_rules_reason() {
        let m = match_of("OL-CMD-ENF", false);
        let mut response = VerdictResponse::deny(
            "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 deny"
        );
    }

    /// 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_deny_renders_the_rules_reason_to_the_agent() {
        let m = match_of("OL-CMD-ENF", false);
        let response = VerdictResponse::deny(
            "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 deny 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"));
    }
}