openlatch-client 0.3.3

//! Session registry + identity assurance (D-29).
//!
//! There is no active-session store in the daemon `AppState` — this builds one.
//! The Mode 1 hook already fires on `SessionStart` and every tool call; the hook
//! side (`daemon::handlers::process_envelope`) stamps a **shared** registry that
//! the boundary listener reads at request time to resolve the attribution triple
//! (`agent_id` / `source` / `agent_session_id`) plus an honest confidence signal
//! (`assurance`).
//!
//! > **B-2 / C-2b.** Nothing session-specific travels on the model request. The
//! > boundary and the hook run in the **same binary**, so the current session is
//! > correlated *in-process* against this shared registry, keyed by the stable
//! > per-install id. `agent_id` + `source` are the platform join keys the hook
//! > materializer resolves against `agents` — carried here verbatim so the
//! > economics row joins the same way the hook stream does.
//!
//! The registry is written by the hook side and read by the boundary side. Both
//! hold the **same** `Arc<SessionRegistry>` (created once in
//! `daemon::serve_with_listener`), so an upsert on the hook path is visible to
//! the very next boundary request.

use std::cmp::Reverse;
use std::collections::hash_map::DefaultHasher;
use std::collections::VecDeque;
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use dashmap::DashMap;

/// How many recent `tool_use_id`s to keep per session for the tool-result join.
///
/// The join only needs to cover the ids a request could still be answering: the
/// agent sends `tool_result` blocks for the turn it just executed, so a handful
/// spans a parallel-tool turn with room to spare. Bounded because this is a
/// per-session allocation that a long conversation would otherwise grow forever.
///
/// Visible to [`super::wire_format`] because it is the request side's bound too:
/// a request may echo a whole conversation's ids, and every one older than the
/// newest few has nothing left here to match. One number, read from both sides.
pub(super) const RECENT_TOOL_USE_IDS: usize = 8;

/// How many recent prompt hashes to keep per session for the prompt join.
///
/// Bounded for the same reason as [`RECENT_TOOL_USE_IDS`], and sized the same:
/// it only has to span the turns a request could still be carrying.
const RECENT_PROMPT_HASHES: usize = 8;

/// Hash prompt text to a `u64` rather than retaining it.
///
/// The cascade only ever asks "is this the same string?", which a hash answers,
/// so the registry never holds prompt content. That keeps C-10 / D-06 / D-17
/// intact: the matching is in-process on content both layers already see, and
/// what we *store* is not the content.
pub fn text_hash(s: &str) -> u64 {
    let mut h = DefaultHasher::new();
    s.trim().hash(&mut h);
    h.finish()
}

/// The stable, opaque, PII-free per-install identifier. In Phase 1 it reuses the
/// existing `agent_id` (`agt_<uuid>`) — the hook writes it into
/// `ANTHROPIC_CUSTOM_HEADERS` at `init`, so it arrives back on every model
/// request as `x-openlatch-install-id`, keying the registry from both sides.
pub type InstallId = String;

/// How long an unrefreshed session stays "active". Past this quiet window an
/// entry is dropped lazily on the next read/write — a stopped agent no longer
/// counts toward the concurrency tie-break.
pub const SESSION_QUIET_WINDOW: Duration = Duration::from_secs(300);

/// Cap on distinct installs the registry tracks, so a pathological caller that
/// forges a fresh `x-openlatch-install-id` per request cannot grow it unbounded.
/// Loopback-only + one install per host makes 1024 generous headroom.
const MAX_TRACKED_INSTALLS: usize = 1024;

/// Identity-assurance — the honest confidence signal (C-7, **three** values, no
/// `asserted`). Wire field `ai.openlatch.session.assurance`; storage column
/// `identity_assurance`. Same value under two names by design.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Assurance {
    /// Exactly one agent session active for this install at request time.
    Attested,
    /// Two or more concurrent — correlation is a most-recent-active best guess.
    Inferred,
    /// No hook session active (agent without the hook, or a non-agent caller).
    Unknown,
}

/// The wire key [`Assurance`] travels under, inside the envelope's `data`.
///
/// Named once because it now has two writers: the boundary writes it on every
/// economics event ([`super::emit`]), and the cloud worker writes it on any
/// event whose session it had to resolve at egress
/// (`core::cloud::worker::stamp_session`). The platform reads one key for both,
/// so the two must not be free to drift apart.
pub const ASSURANCE_KEY: &str = "ai.openlatch.session.assurance";

impl Assurance {
    /// Frozen wire string.
    pub fn as_str(&self) -> &'static str {
        match self {
            Assurance::Attested => "attested",
            Assurance::Inferred => "inferred",
            Assurance::Unknown => "unknown",
        }
    }
}

/// One live agent session, as last seen by the hook side.
#[derive(Clone, Debug)]
pub struct SessionActivity {
    /// The attribution resolution key (C-2b) — the same `agent_id` the hook emits.
    pub agent_id: String,
    /// The agent platform (`claude-code`) — first component of the `agents`
    /// unique key. Carried verbatim from the hook envelope so the join matches.
    pub source: String,
    /// The agent session id (the hook envelope `subject`).
    pub session_id: String,
    /// Last time the hook refreshed this session.
    pub last_seen: Instant,
    /// Recent `tool_use_id`s this session executed, newest first, capped at
    /// [`RECENT_TOOL_USE_IDS`]. The agent's next model request echoes these back
    /// in its `tool_result` blocks, which is what makes selector 1 an **exact
    /// join** rather than a similarity guess.
    pub recent_tool_use_ids: VecDeque<String>,
    /// Hashes of the recent user prompts this session submitted, newest first,
    /// capped at [`RECENT_PROMPT_HASHES`].
    ///
    /// A bounded **set**, not the single opening-turn anchor this began as. That
    /// anchor could only be the session's true first prompt if the entry already
    /// existed when that prompt was typed — and this registry is in-memory, dies
    /// with the daemon, and evicts an entry after [`SESSION_QUIET_WINDOW`] of
    /// silence. Every rebirth therefore re-anchored a long-running session to a
    /// *mid-conversation* turn while the request kept offering `messages[0]`, so
    /// the two sides compared different strings and selector 2 could not match
    /// again for the rest of that session's life. Keeping the recent turns
    /// instead means the join survives a restart, a quiet gap, and a compaction:
    /// the request carries those same turns.
    pub recent_prompt_hashes: VecDeque<u64>,
}

/// What the hook side learned about a session from one envelope.
///
/// A struct rather than more positional parameters: every field is optional and
/// most events carry only some of them, so the call site stays readable as the
/// set grows.
#[derive(Clone, Debug, Default)]
pub struct SessionSignals<'a> {
    /// `tool_use_id` from a `pre_tool_use` / `post_tool_use` envelope.
    pub tool_use_id: Option<&'a str>,
    /// The prompt text from a `user_prompt_submit` envelope. Hashed on write.
    pub prompt: Option<&'a str>,
}

/// What the boundary side can see in one model request, for the cascade.
///
/// Empty is meaningful and always valid: the metadata-only path has no body to
/// read, so it passes `default()` and the cascade falls straight through to
/// most-recently-active — the pre-cascade behaviour.
#[derive(Clone, Debug, Default)]
pub struct RequestSignals {
    /// The session id the request **declares about itself**, when the agent
    /// carries one (Claude Code puts it in `metadata.user_id`).
    ///
    /// This is not a signal to match on — it is the answer. Every heuristic
    /// below exists because the design took B-2 ("`ANTHROPIC_CUSTOM_HEADERS` is
    /// static, so the request carries no session id") to mean no session id was
    /// available anywhere. B-2 is right about *headers*; the body was never
    /// checked, and Claude Code has been naming its own session in it all along.
    pub declared_session_id: Option<String>,
    /// `tool_use_id`s carried by `tool_result` blocks in this request.
    pub tool_use_ids: Vec<String>,
    /// Candidate hashes of this request's user turns — a match of **any** of
    /// them against **any** hash the session recorded satisfies selector 2.
    ///
    /// Several candidates per turn, because the hook and the request do not see
    /// the same string. The hook stores the hash of the prompt the user typed; by
    /// the time that turn reaches the provider the agent may have appended its
    /// own text blocks (Claude Code appends system reminders), so a hash of the
    /// whole concatenated turn would never equal the hook's. Offering the first
    /// block alone *and* the full concatenation lets the exact-equality test hold
    /// in both shapes without ever storing the text to compare loosely.
    ///
    /// Several turns, because the opening turn is the one turn a registry entry
    /// born mid-conversation is guaranteed NOT to hold. The recent turns are what
    /// both sides reliably share.
    pub prompt_hashes: Vec<u64>,
}

impl RequestSignals {
    /// True when nothing content-derived is present — the cascade then has
    /// nothing to match on and the fallback decides.
    fn is_empty(&self) -> bool {
        self.declared_session_id.is_none()
            && self.tool_use_ids.is_empty()
            && self.prompt_hashes.is_empty()
    }
}

/// Which selector placed the request. Carried so the assurance mapping lives in
/// exactly one place instead of being re-derived at each return site.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Selector {
    /// Exactly one session was live — no cascade needed, and no selector could
    /// change the answer.
    OnlyLive,
    /// Selector 0: the request named its own session, and that session is live.
    /// Not a heuristic — the agent's own statement of which session it is.
    SessionId,
    /// Selector 0, race arm: the request named a session the registry does not
    /// hold yet. The agent's **first** model call races the `SessionStart` hook
    /// that registers it, so this is not an error — it is the one case every
    /// other selector gets wrong, since the session has no history to match on.
    SessionIdUnregistered,
    /// Selector 1: a `tool_result` in the request echoes an id this session ran.
    ToolResult,
    /// Selector 2: a user turn in the request is one this session submitted.
    Prompt,
    /// Selector 3: the fallback — whichever session's hook fired most recently.
    /// This is the pre-cascade behaviour unchanged — D-29 selector 3.
    MostRecentActive,
}

impl Selector {
    /// Stable label for bench reporting.
    pub fn as_str(self) -> &'static str {
        match self {
            Selector::OnlyLive => "only_live",
            Selector::SessionId => "session_id",
            Selector::SessionIdUnregistered => "session_id_unregistered",
            Selector::ToolResult => "tool_result",
            Selector::Prompt => "prompt",
            Selector::MostRecentActive => "most_recent_active",
        }
    }

    /// Every variant, in declaration order. The counter array is indexed by
    /// position here, so the two must not drift — [`Selector::index`] is the
    /// only thing that maps between them and this is its source of truth.
    pub const ALL: [Selector; 6] = [
        Selector::OnlyLive,
        Selector::SessionId,
        Selector::SessionIdUnregistered,
        Selector::ToolResult,
        Selector::Prompt,
        Selector::MostRecentActive,
    ];

    /// Position in [`Selector::ALL`] — the [`SELECTOR_WINS`] index.
    fn index(self) -> usize {
        match self {
            Selector::OnlyLive => 0,
            Selector::SessionId => 1,
            Selector::SessionIdUnregistered => 2,
            Selector::ToolResult => 3,
            Selector::Prompt => 4,
            Selector::MostRecentActive => 5,
        }
    }

    /// The honest confidence this selector earns, within the frozen three-value
    /// enum (C-7). Content selectors identify a session; the others pick one.
    fn assurance(self) -> Assurance {
        match self {
            // An exact echo of an id this session ran, or of a prompt it
            // submitted, identifies the session as well as a single live session
            // does — there is nothing left to guess.
            // A request that names its own session is the strongest evidence
            // available — stronger than "only one was live", which is an absence
            // of alternatives rather than an identification.
            Selector::OnlyLive
            | Selector::SessionId
            | Selector::SessionIdUnregistered
            | Selector::ToolResult
            | Selector::Prompt => Assurance::Attested,
            // A repo narrows; it does not identify. Two sessions in one checkout
            // both match, and the fallback is a guess by construction.
            Selector::MostRecentActive => Assurance::Inferred,
        }
    }
}

impl SessionActivity {
    /// Fold one envelope's signals in, leaving untouched anything it did not carry.
    fn apply(&mut self, signals: &SessionSignals<'_>) {
        if let Some(id) = signals.tool_use_id.filter(|s| !s.is_empty()) {
            // Newest first, and never the same id twice — a pre/post pair for
            // one tool call would otherwise spend two of the eight slots.
            if !self.recent_tool_use_ids.iter().any(|k| k == id) {
                self.recent_tool_use_ids.push_front(id.to_string());
                self.recent_tool_use_ids.truncate(RECENT_TOOL_USE_IDS);
            }
        }
        if let Some(p) = signals.prompt.filter(|s| !s.trim().is_empty()) {
            // Newest first, deduped and bounded — the same shape as the tool-use
            // ids above, and for the same reason: the request echoes whichever of
            // these turns its history still holds, so a *set* joins where the
            // single opening-turn anchor this replaced could not.
            let h = text_hash(p);
            if !self.recent_prompt_hashes.contains(&h) {
                self.recent_prompt_hashes.push_front(h);
                self.recent_prompt_hashes.truncate(RECENT_PROMPT_HASHES);
            }
        }
    }
}

/// Shared active-session store. `DashMap` (already a dependency) gives lock-free
/// concurrent access from the hook write path and the boundary read path.
#[derive(Default)]
pub struct SessionRegistry {
    active: DashMap<InstallId, Vec<SessionActivity>>,
}

impl SessionRegistry {
    /// Hook side: record (or refresh) a session for this install. Called on
    /// `SessionStart` and every tool-call hook. Stale siblings are pruned in
    /// passing so the concurrency count stays honest.
    pub fn upsert(&self, install_id: &str, agent_id: &str, source: &str, session_id: &str) {
        self.upsert_signals(
            install_id,
            agent_id,
            source,
            session_id,
            &SessionSignals::default(),
        );
    }

    /// As [`SessionRegistry::upsert`], additionally folding in whatever the
    /// envelope told us about this session for the cascade.
    ///
    /// Signals are **merged, never cleared**: an envelope that carries no
    /// `tool_use_id` must not erase the ids a previous one recorded, or the very
    /// next request would lose the join it was about to match on.
    pub fn upsert_signals(
        &self,
        install_id: &str,
        agent_id: &str,
        source: &str,
        session_id: &str,
        signals: &SessionSignals<'_>,
    ) {
        // Bound growth: if we are at the install cap and this is a brand-new
        // install key, drop it rather than grow unbounded (loopback, one install
        // per host — this only trips under a forging caller).
        if !self.active.contains_key(install_id) && self.active.len() >= MAX_TRACKED_INSTALLS {
            self.evict_empty();
            if self.active.len() >= MAX_TRACKED_INSTALLS {
                return;
            }
        }

        let now = Instant::now();
        let mut entry = self.active.entry(install_id.to_string()).or_default();
        entry.retain(|a| now.duration_since(a.last_seen) < SESSION_QUIET_WINDOW);
        if let Some(existing) = entry.iter_mut().find(|a| a.session_id == session_id) {
            existing.last_seen = now;
            existing.agent_id = agent_id.to_string();
            existing.source = source.to_string();
            existing.apply(signals);
        } else {
            let mut fresh = SessionActivity {
                agent_id: agent_id.to_string(),
                source: source.to_string(),
                session_id: session_id.to_string(),
                last_seen: now,
                recent_tool_use_ids: VecDeque::new(),
                recent_prompt_hashes: VecDeque::new(),
            };
            fresh.apply(signals);
            entry.push(fresh);
        }
    }

    /// Drop install keys whose session vectors have gone fully stale/empty.
    fn evict_empty(&self) {
        let now = Instant::now();
        self.active.retain(|_, v| {
            v.retain(|a| now.duration_since(a.last_seen) < SESSION_QUIET_WINDOW);
            !v.is_empty()
        });
    }

    /// The fresh (non-stale) sessions for an install, dropping expired entries.
    fn fresh(&self, install_id: &str) -> Vec<SessionActivity> {
        let now = Instant::now();
        match self.active.get(install_id) {
            Some(v) => v
                .iter()
                .filter(|a| now.duration_since(a.last_seen) < SESSION_QUIET_WINDOW)
                .cloned()
                .collect(),
            None => Vec::new(),
        }
    }
}

/// The resolved attribution triple + assurance for one request.
#[derive(Clone, Debug)]
pub struct Resolved {
    pub agent_id: Option<String>,
    pub source: Option<String>,
    pub session_id: Option<String>,
    pub assurance: Assurance,
}

impl Resolved {
    /// The unattributed resolution — no hook session for this install: no
    /// identifiers, `unknown` assurance. Still forwarded, just Unattributed.
    pub fn unknown() -> Self {
        Resolved {
            agent_id: None,
            source: None,
            session_id: None,
            assurance: Assurance::Unknown,
        }
    }
}

/// Boundary side (D-29): resolve the attribution triple + assurance for an
/// install at request time.
///
/// - **1 active** → `attested`, that session's identifiers.
/// - **0 active** → `unknown`, no identifiers (still forwarded, Unattributed).
/// - **≥2 active** → `inferred`, the **most-recently-active** session (flagged).
pub fn resolve_session(reg: &SessionRegistry, install: &str) -> Resolved {
    resolve_session_with(reg, install, &RequestSignals::default())
}

/// The cascade entry point: resolve the attribution triple + assurance, using
/// the request's own content to pick between concurrent sessions.
///
/// **Empty signals are the off switch, and the only one.** `RequestSignals`
/// carrying nothing reproduces [`resolve_session`] exactly — no selector can
/// match, so the most-recently-active pick decides and is flagged `inferred`.
/// That is what the metadata-only path passes, having no body to read, and it is
/// what the bench passes for its pre-cascade control. There is no separate flag:
/// a request whose content identifies a session is attributed to that session,
/// always.
pub fn resolve_session_with(
    reg: &SessionRegistry,
    install: &str,
    signals: &RequestSignals,
) -> Resolved {
    resolve_session_detailed(reg, install, signals).0
}

/// Process-wide tally of which selector decided, indexed by [`Selector::index`].
///
/// Until this existed the boundary emitted the *assurance* of every resolution
/// but never which selector produced it, so the cascade was unfalsifiable in
/// production: an `inferred` row could not be told apart from a `tool_result`
/// join that had quietly stopped matching, and the only evidence the cascade
/// worked at all came from the bench. These counters are what make a live
/// deployment measurable — they are read by `GET /admin/boundary/status`.
///
/// Process-wide statics rather than [`BoundaryState`] fields, mirroring
/// `PASS_THROUGH_FAILURES`: one listener per host by construction, and a
/// `Relaxed` add on an already-hot path costs nothing.
///
/// [`BoundaryState`]: super::BoundaryState
static SELECTOR_WINS: [AtomicU64; 6] = [
    AtomicU64::new(0),
    AtomicU64::new(0),
    AtomicU64::new(0),
    AtomicU64::new(0),
    AtomicU64::new(0),
    AtomicU64::new(0),
];

/// Resolutions that found **no** live session at all (`unknown` assurance).
///
/// Not a [`Selector`] variant because no selector ran: the live set was empty
/// before the cascade was consulted. Counted separately so a host that is
/// simply not being hooked reads differently from one whose selectors are
/// failing to match.
static NO_LIVE_SESSION: AtomicU64 = AtomicU64::new(0);

/// Snapshot the selector tally: `(label, wins)` per selector, plus the
/// `no_live_session` count under its own label.
pub fn selector_wins() -> Vec<(&'static str, u64)> {
    let mut out: Vec<(&'static str, u64)> = Selector::ALL
        .iter()
        .map(|s| (s.as_str(), SELECTOR_WINS[s.index()].load(Ordering::Relaxed)))
        .collect();
    out.push(("no_live_session", NO_LIVE_SESSION.load(Ordering::Relaxed)));
    out
}

/// As [`resolve_session_with`], also reporting **which** selector decided.
///
/// The selector is not on [`Resolved`] because nothing on the request path needs
/// it — the assurance already carries the honesty signal. The attribution bench
/// needs it to report a per-selector win rate, and `boundary status` renders the
/// same tally so the cascade stays falsifiable in production.
pub fn resolve_session_detailed(
    reg: &SessionRegistry,
    install: &str,
    signals: &RequestSignals,
) -> (Resolved, Option<Selector>) {
    let active = reg.fresh(install);

    // Selector 0, and it runs before the live-set checks on purpose.
    //
    // Everything below reasons about *which of the live sessions* this request
    // belongs to. When the request names its own session that question is
    // already answered, and — critically — the answer holds even when the
    // registry has never heard of that session. An agent's FIRST model call
    // races the `SessionStart` hook that would register it, which is exactly
    // when the live set is either empty or contains only other people's
    // sessions. Deferring this check until after those arms would hand the one
    // case no heuristic can win back to the heuristics.
    // Trimmed, and blank-rejected: the proxy already normalises what it parses,
    // but this is the API every other caller reaches too, and a whitespace-only
    // id would otherwise become a session label that matches nothing and reads
    // as real.
    if let Some(sid) = signals
        .declared_session_id
        .as_deref()
        .map(str::trim)
        .filter(|s| !s.is_empty())
    {
        let selector = match active.iter().find(|a| a.session_id == sid) {
            // Live and named: take the registry's row, so `source` and
            // `agent_id` come from the hook that actually observed it.
            Some(hit) => {
                SELECTOR_WINS[Selector::SessionId.index()].fetch_add(1, Ordering::Relaxed);
                return (
                    resolved(hit, Selector::SessionId),
                    Some(Selector::SessionId),
                );
            }
            // Named but not registered yet. Attribute anyway: the id is the
            // agent's own claim about itself and needs no corroboration to be
            // the right label. `source` is left None rather than borrowed from
            // an unrelated session — `emit` already falls back to the default
            // agent platform, and inventing a neighbour's value would be a
            // guess dressed as a fact.
            None => Selector::SessionIdUnregistered,
        };
        SELECTOR_WINS[selector.index()].fetch_add(1, Ordering::Relaxed);
        return (
            Resolved {
                agent_id: Some(install.to_string()),
                source: None,
                session_id: Some(sid.to_string()),
                assurance: selector.assurance(),
            },
            Some(selector),
        );
    }

    if active.is_empty() {
        NO_LIVE_SESSION.fetch_add(1, Ordering::Relaxed);
        return (Resolved::unknown(), None);
    }
    // One live session is not a guess: there is nothing to disambiguate and no
    // selector could change the answer.
    if active.len() == 1 {
        SELECTOR_WINS[Selector::OnlyLive.index()].fetch_add(1, Ordering::Relaxed);
        return (
            resolved(&active[0], Selector::OnlyLive),
            Some(Selector::OnlyLive),
        );
    }

    let (winner, selector) = if signals.is_empty() {
        // Nothing content-derived to match on — the metadata-only path, or a
        // body no selector could read. The fallback decides, and says so.
        (most_recent(active), Selector::MostRecentActive)
    } else {
        cascade_pick(active, signals)
    };
    SELECTOR_WINS[selector.index()].fetch_add(1, Ordering::Relaxed);
    (resolved(&winner, selector), Some(selector))
}

/// Run the ordered selectors over the live set.
///
/// Fall-through per the cascade design: a selector matching **exactly one**
/// session wins; one matching **several** narrows the candidate set and hands
/// the tie to the next selector; one matching **none** leaves the set untouched.
/// Selector 4 is always decisive, so a non-empty live set always resolves.
fn cascade_pick(
    active: Vec<SessionActivity>,
    signals: &RequestSignals,
) -> (SessionActivity, Selector) {
    let mut pool = active;

    // 1 — tool-result join. Exact, and durable: unlike the first prompt it
    // survives compaction and any history trim, because the agent must echo the
    // id of the tool call it is answering.
    if !signals.tool_use_ids.is_empty() {
        let matched = narrow(&pool, |a| {
            signals
                .tool_use_ids
                .iter()
                .any(|id| a.recent_tool_use_ids.iter().any(|k| k == id))
        });
        match matched.len() {
            1 => {
                return (
                    matched.into_iter().next().expect("len checked"),
                    Selector::ToolResult,
                )
            }
            n if n > 1 => pool = matched,
            _ => {}
        }
    }

    // 2 — prompt join. Exact, and no longer pinned to the opening turn: both
    // sides keep a bounded set of recent turns, so an entry born mid-conversation
    // still intersects the history the request carries.
    if !signals.prompt_hashes.is_empty() {
        let matched = narrow(&pool, |a| {
            a.recent_prompt_hashes
                .iter()
                .any(|h| signals.prompt_hashes.contains(h))
        });
        match matched.len() {
            1 => {
                return (
                    matched.into_iter().next().expect("len checked"),
                    Selector::Prompt,
                )
            }
            n if n > 1 => pool = matched,
            _ => {}
        }
    }

    // A repo / cwd selector was designed here and deliberately left out. It
    // matched a session whose working directory prefixes a path in the request,
    // which is NOT proof of ownership: an agent reading a file under another
    // session's checkout would be handed to that session, and the fallback below
    // might have been right. It is the only selector that could attribute WORSE
    // than the behaviour it replaces, so the cascade keeps only the two exact
    // matches above plus the fallback — with those, it cannot regress.
    //
    // 3 — the fallback, applied to whatever the cascade narrowed to.
    (most_recent(pool), Selector::MostRecentActive)
}

/// The subset of `pool` matching `pred`.
fn narrow(
    pool: &[SessionActivity],
    pred: impl Fn(&SessionActivity) -> bool,
) -> Vec<SessionActivity> {
    pool.iter().filter(|a| pred(a)).cloned().collect()
}

/// Most-recently-active of a non-empty set (the pre-cascade tie-break).
fn most_recent(mut pool: Vec<SessionActivity>) -> SessionActivity {
    pool.sort_by_key(|a| Reverse(a.last_seen));
    pool.remove(0)
}

/// Build the resolution, mapping the winning selector to its assurance.
fn resolved(a: &SessionActivity, selector: Selector) -> Resolved {
    Resolved {
        agent_id: Some(a.agent_id.clone()),
        source: Some(a.source.clone()),
        session_id: Some(a.session_id.clone()),
        assurance: selector.assurance(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn zero_active_resolves_unknown() {
        // C-8b: no hook session for this install → unknown, no identifiers.
        let reg = SessionRegistry::default();
        let r = resolve_session(&reg, "agt_absent");
        assert_eq!(r.assurance, Assurance::Unknown);
        assert!(r.agent_id.is_none() && r.session_id.is_none() && r.source.is_none());
    }

    #[test]
    fn one_active_resolves_attested_with_identifiers() {
        let reg = SessionRegistry::default();
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        let r = resolve_session(&reg, "agt_1");
        assert_eq!(r.assurance, Assurance::Attested);
        assert_eq!(r.session_id.as_deref(), Some("sess_a"));
        assert_eq!(r.agent_id.as_deref(), Some("agt_1"));
        assert_eq!(r.source.as_deref(), Some("claude-code"));
    }

    #[test]
    fn two_concurrent_resolve_inferred_most_recent() {
        let reg = SessionRegistry::default();
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_old");
        std::thread::sleep(Duration::from_millis(5));
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_new");
        let r = resolve_session(&reg, "agt_1");
        assert_eq!(r.assurance, Assurance::Inferred);
        // Most-recently-active session wins the tie-break.
        assert_eq!(r.session_id.as_deref(), Some("sess_new"));
    }

    #[test]
    fn refresh_updates_last_seen_not_count() {
        let reg = SessionRegistry::default();
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        reg.upsert("agt_1", "agt_1", "claude-code", "sess_a");
        // Same session refreshed → still one active → attested, not inferred.
        let r = resolve_session(&reg, "agt_1");
        assert_eq!(r.assurance, Assurance::Attested);
    }

    #[test]
    fn assurance_wire_strings_are_the_frozen_three() {
        assert_eq!(Assurance::Attested.as_str(), "attested");
        assert_eq!(Assurance::Inferred.as_str(), "inferred");
        assert_eq!(Assurance::Unknown.as_str(), "unknown");
    }

    // ---- cascade -------------------------------------------------------

    /// Two live sessions, `a` older than `b`, with the signals each recorded.
    fn two_live() -> SessionRegistry {
        let reg = SessionRegistry::default();
        reg.upsert_signals(
            "i",
            "i",
            "claude-code",
            "sess_a",
            &SessionSignals {
                tool_use_id: Some("toolu_aaa"),
                prompt: Some("build the parser"),
            },
        );
        std::thread::sleep(Duration::from_millis(5));
        reg.upsert_signals(
            "i",
            "i",
            "claude-code",
            "sess_b",
            &SessionSignals {
                tool_use_id: Some("toolu_bbb"),
                prompt: Some("write the docs"),
            },
        );
        reg
    }

    #[test]
    fn no_signals_at_all_keeps_most_recent_active() {
        // The metadata-only path, and the only remaining way to get the
        // pre-cascade pick: nothing content-derived to match on, so recency
        // decides and says out loud that it guessed.
        let reg = two_live();
        let r = resolve_session_with(&reg, "i", &RequestSignals::default());
        assert_eq!(r.session_id.as_deref(), Some("sess_b"));
        assert_eq!(r.assurance, Assurance::Inferred);
    }

    #[test]
    fn tool_result_join_beats_recency_and_attests() {
        // The heart of it: the older session ran this tool call, so it owns the
        // request that answers it — regardless of who moved most recently.
        let reg = two_live();
        let signals = RequestSignals {
            tool_use_ids: vec!["toolu_aaa".into()],
            ..Default::default()
        };
        let r = resolve_session_with(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_a"));
        assert_eq!(r.assurance, Assurance::Attested);
    }

    #[test]
    fn prompt_match_attests() {
        let reg = two_live();
        let signals = RequestSignals {
            prompt_hashes: vec![text_hash("build the parser")],
            ..Default::default()
        };
        let r = resolve_session_with(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_a"));
        assert_eq!(r.assurance, Assurance::Attested);
    }

    #[test]
    fn signals_that_match_nothing_fall_through_to_most_recent() {
        let reg = two_live();
        let signals = RequestSignals {
            declared_session_id: None,
            tool_use_ids: vec!["toolu_zzz".into()],
            prompt_hashes: vec![text_hash("something else entirely")],
        };
        let r = resolve_session_with(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_b"));
        assert_eq!(r.assurance, Assurance::Inferred);
    }

    #[test]
    fn a_tie_narrows_the_pool_before_the_fallback_decides() {
        // Two sessions were opened with the SAME prompt, so selector 2 matches
        // both; a third (the most recent overall) opened with a different one.
        // The tie must narrow to the first two and let the fallback decide WITHIN
        // that pool — never hand the request to the newer outsider.
        let reg = SessionRegistry::default();
        for sess in ["sess_a", "sess_b"] {
            reg.upsert_signals(
                "i",
                "i",
                "claude-code",
                sess,
                &SessionSignals {
                    prompt: Some("same opening prompt"),
                    ..Default::default()
                },
            );
            std::thread::sleep(Duration::from_millis(5));
        }
        reg.upsert_signals(
            "i",
            "i",
            "claude-code",
            "sess_outsider",
            &SessionSignals {
                prompt: Some("a different opening"),
                ..Default::default()
            },
        );
        let signals = RequestSignals {
            prompt_hashes: vec![text_hash("same opening prompt")],
            ..Default::default()
        };
        let r = resolve_session_with(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_b"));
        // A tie resolved by recency is a guess, not an identification.
        assert_eq!(r.assurance, Assurance::Inferred);
    }

    #[test]
    fn empty_signals_with_cascade_on_are_the_metadata_only_path() {
        // `observe_request_metadata_only` has no body to read. That must resolve,
        // and must resolve honestly as a guess.
        let reg = two_live();
        let r = resolve_session_with(&reg, "i", &RequestSignals::default());
        assert_eq!(r.session_id.as_deref(), Some("sess_b"));
        assert_eq!(r.assurance, Assurance::Inferred);
    }

    #[test]
    fn one_live_session_attests_with_the_cascade_on() {
        let reg = SessionRegistry::default();
        reg.upsert("i", "i", "claude-code", "sess_only");
        let signals = RequestSignals {
            tool_use_ids: vec!["toolu_unrelated".into()],
            ..Default::default()
        };
        let r = resolve_session_with(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_only"));
        assert_eq!(r.assurance, Assurance::Attested);
    }

    #[test]
    fn no_live_session_is_unknown_whatever_the_signals() {
        let signals = RequestSignals {
            tool_use_ids: vec!["toolu_aaa".into()],
            ..Default::default()
        };
        let r = resolve_session_with(&SessionRegistry::default(), "i", &signals);
        assert_eq!(r.assurance, Assurance::Unknown);
        assert!(r.session_id.is_none());
    }

    // ---- selector 0: the request names its own session -------------------

    #[test]
    fn a_request_that_names_its_session_beats_every_heuristic() {
        // `sess_a` is older AND the request carries a tool_use_id belonging to
        // nobody. Neither recency nor content should get a vote once the request
        // has stated which session it is.
        let reg = two_live();
        let signals = RequestSignals {
            declared_session_id: Some("sess_a".into()),
            tool_use_ids: vec!["toolu_bbb".into()],
            ..Default::default()
        };
        let (r, sel) = resolve_session_detailed(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_a"));
        assert_eq!(sel, Some(Selector::SessionId));
        assert_eq!(r.assurance, Assurance::Attested);
        // The registry row supplies source/agent_id when it has one.
        assert_eq!(r.source.as_deref(), Some("claude-code"));
    }

    #[test]
    fn a_named_session_the_registry_has_not_seen_yet_is_still_attributed() {
        // THE field failure this fixes. An agent's first model call races the
        // `SessionStart` hook, so the real session is not in the registry while
        // stale ones still are — every heuristic then picks a stale one. The
        // declared id is right regardless of what the registry knows.
        let reg = two_live();
        let signals = RequestSignals {
            declared_session_id: Some("sess_brand_new".into()),
            ..Default::default()
        };
        let (r, sel) = resolve_session_detailed(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_brand_new"));
        assert_eq!(sel, Some(Selector::SessionIdUnregistered));
        assert_eq!(r.assurance, Assurance::Attested);
        assert_eq!(r.agent_id.as_deref(), Some("i"));
        // Never borrowed from a neighbouring session — `emit` defaults it.
        assert!(
            r.source.is_none(),
            "a source we do not know must not be invented"
        );
    }

    #[test]
    fn a_named_session_resolves_even_with_an_empty_registry() {
        // The same race, at its extreme: the very first request of the very
        // first session on a freshly started daemon. Pre-selector-0 this was
        // `unknown` — Unattributed — despite the request saying who it was.
        let signals = RequestSignals {
            declared_session_id: Some("sess_first".into()),
            ..Default::default()
        };
        let (r, sel) = resolve_session_detailed(&SessionRegistry::default(), "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_first"));
        assert_eq!(sel, Some(Selector::SessionIdUnregistered));
        assert_eq!(r.assurance, Assurance::Attested);
    }

    #[test]
    fn the_zombie_session_misattribution_is_fixed() {
        // Reproduces the two rows observed in the field on 2026-08-24: sessions
        // dead for 40 minutes were kept in the live set by daemon-generated
        // `config_change` envelopes, and the first request of a genuinely new
        // session was handed to one of them.
        let reg = SessionRegistry::default();
        for zombie in ["6693e5f6", "a5a39553", "92129c1b"] {
            reg.upsert("i", "i", "claude-code", zombie);
        }
        let signals = RequestSignals {
            declared_session_id: Some("5c7d9833".into()),
            ..Default::default()
        };
        // Without the request's own claim the bug reproduces: some zombie wins.
        // This is not a hypothetical control — it is exactly what the
        // metadata-only path passes when the body could not be read.
        let blind = resolve_session_with(&reg, "i", &RequestSignals::default());
        assert_ne!(blind.session_id.as_deref(), Some("5c7d9833"));
        assert_eq!(blind.assurance, Assurance::Inferred);
        // With it, the request is believed.
        let seen = resolve_session_with(&reg, "i", &signals);
        assert_eq!(seen.session_id.as_deref(), Some("5c7d9833"));
        assert_eq!(seen.assurance, Assurance::Attested);
    }

    #[test]
    fn an_empty_declared_session_id_falls_through_rather_than_labelling_nothing() {
        let reg = two_live();
        let signals = RequestSignals {
            declared_session_id: Some("   ".into()),
            tool_use_ids: vec!["toolu_aaa".into()],
            ..Default::default()
        };
        let (r, sel) = resolve_session_detailed(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_a"));
        assert_eq!(sel, Some(Selector::ToolResult));
    }

    // ---- signal bookkeeping --------------------------------------------

    /// The counters are what make the cascade falsifiable in production, so they
    /// have to actually move when a selector decides.
    ///
    /// Asserted as a **delta**, never an absolute: [`SELECTOR_WINS`] is
    /// process-wide and the rest of this module's tests resolve sessions on
    /// other threads, so any exact expectation here would be flaky by
    /// construction. Other tests can only ever add, so `after > before` holds.
    #[test]
    fn a_resolution_is_counted_under_the_selector_that_decided() {
        let win = |sel: Selector| -> u64 {
            selector_wins()
                .into_iter()
                .find(|(k, _)| *k == sel.as_str())
                .map(|(_, v)| v)
                .expect("every selector is reported")
        };
        let unattributed = || -> u64 {
            selector_wins()
                .into_iter()
                .find(|(k, _)| *k == "no_live_session")
                .map(|(_, v)| v)
                .expect("no_live_session is reported")
        };

        // A tool-result join.
        let before = win(Selector::ToolResult);
        let reg = two_live();
        let signals = RequestSignals {
            tool_use_ids: vec!["toolu_aaa".to_string()],
            ..Default::default()
        };
        let (_, sel) = resolve_session_detailed(&reg, "i", &signals);
        assert_eq!(sel, Some(Selector::ToolResult));
        assert!(
            win(Selector::ToolResult) > before,
            "a tool_result win must be counted"
        );

        // An empty live set is counted apart from every selector.
        let before_unattributed = unattributed();
        let empty = SessionRegistry::default();
        let (r, sel) = resolve_session_detailed(&empty, "nobody", &signals);
        assert_eq!(sel, None);
        assert_eq!(r.assurance, Assurance::Unknown);
        assert!(
            unattributed() > before_unattributed,
            "a resolution with no live session must be counted as such, not as a selector win"
        );
    }

    #[test]
    fn prompt_hashes_accumulate_newest_first_and_are_bounded() {
        let reg = SessionRegistry::default();
        // Re-submitting the same text must not spend two slots.
        for p in ["same turn", "same turn"] {
            reg.upsert_signals(
                "i",
                "i",
                "claude-code",
                "s",
                &SessionSignals {
                    prompt: Some(p),
                    ..Default::default()
                },
            );
        }
        assert_eq!(reg.fresh("i")[0].recent_prompt_hashes.len(), 1);

        for n in 0..(RECENT_PROMPT_HASHES + 5) {
            reg.upsert_signals(
                "i",
                "i",
                "claude-code",
                "s",
                &SessionSignals {
                    prompt: Some(&format!("turn {n}")),
                    ..Default::default()
                },
            );
        }
        let live = reg.fresh("i");
        assert_eq!(live[0].recent_prompt_hashes.len(), RECENT_PROMPT_HASHES);
        // Newest first — the oldest turns were the ones dropped.
        assert_eq!(
            live[0].recent_prompt_hashes.front().copied(),
            Some(text_hash(&format!("turn {}", RECENT_PROMPT_HASHES + 4)))
        );
    }

    #[test]
    fn an_entry_born_mid_conversation_still_joins() {
        // The production failure this selector was rebuilt for. The daemon
        // started ~13 minutes into a live conversation, so the entry never saw
        // the opening turn and anchored on a mid-conversation prompt — while
        // every request kept carrying `messages[0]`. With a single anchor the two
        // sides compared strings that could never be equal, and selector 2 was
        // dead for the rest of that session's life: every request that was not
        // answering a tool call fell through to most-recently-active and was
        // stamped onto whichever session had moved last.
        let reg = SessionRegistry::default();
        for (sess, witnessed) in [
            ("sess_a", "waht is session Id"),
            ("sess_b", "write the docs"),
        ] {
            reg.upsert_signals(
                "i",
                "i",
                "claude-code",
                sess,
                &SessionSignals {
                    prompt: Some(witnessed),
                    ..Default::default()
                },
            );
            std::thread::sleep(Duration::from_millis(5));
        }
        // `sess_b` acted last, so the fallback would hand this request to it.
        let signals = RequestSignals {
            // No declared id either: this test is about the PROMPT join, which
            // is what an agent that does not name its own session relies on.
            declared_session_id: None,
            // No `tool_result`: a fresh user turn echoes nothing, so selector 1
            // cannot fire and selector 2 is all that stands before the fallback.
            tool_use_ids: Vec::new(),
            // What the request carries: the opening turn the registry never saw,
            // plus the later turn it did.
            prompt_hashes: vec![
                text_hash("hello this a session without a cache"),
                text_hash("waht is session Id"),
            ],
        };
        let r = resolve_session_with(&reg, "i", &signals);
        assert_eq!(r.session_id.as_deref(), Some("sess_a"));
        assert_eq!(r.assurance, Assurance::Attested);

        // And the counterfactual, so the old shape cannot quietly come back: with
        // the opening turn as the ONLY candidate — what this offered before —
        // nothing intersects what the entry witnessed, and the request lands on
        // whichever session moved last. That is the misattribution, reproduced.
        let opening_only = RequestSignals {
            declared_session_id: None,
            tool_use_ids: Vec::new(),
            prompt_hashes: vec![text_hash("hello this a session without a cache")],
        };
        let r = resolve_session_with(&reg, "i", &opening_only);
        assert_eq!(r.session_id.as_deref(), Some("sess_b"));
        assert_eq!(r.assurance, Assurance::Inferred);
    }

    #[test]
    fn tool_use_ids_are_deduped_newest_first_and_bounded() {
        let reg = SessionRegistry::default();
        // A pre/post pair for one call must not spend two slots.
        for id in ["toolu_1", "toolu_1"] {
            reg.upsert_signals(
                "i",
                "i",
                "claude-code",
                "s",
                &SessionSignals {
                    tool_use_id: Some(id),
                    ..Default::default()
                },
            );
        }
        assert_eq!(reg.fresh("i")[0].recent_tool_use_ids.len(), 1);

        for n in 2..(RECENT_TOOL_USE_IDS + 5) {
            reg.upsert_signals(
                "i",
                "i",
                "claude-code",
                "s",
                &SessionSignals {
                    tool_use_id: Some(&format!("toolu_{n}")),
                    ..Default::default()
                },
            );
        }
        let live = reg.fresh("i");
        assert_eq!(live[0].recent_tool_use_ids.len(), RECENT_TOOL_USE_IDS);
        // Newest first — the oldest ids were the ones dropped.
        assert_eq!(
            live[0].recent_tool_use_ids.front().map(String::as_str),
            Some(format!("toolu_{}", RECENT_TOOL_USE_IDS + 4).as_str())
        );
    }

    #[test]
    fn a_signal_free_envelope_does_not_clear_what_an_earlier_one_recorded() {
        // `session_end` and friends carry no signals; they must not erase the
        // join the very next request is about to match on.
        let reg = SessionRegistry::default();
        reg.upsert_signals(
            "i",
            "i",
            "claude-code",
            "s",
            &SessionSignals {
                tool_use_id: Some("toolu_keep"),
                prompt: Some("keep me"),
            },
        );
        reg.upsert("i", "i", "claude-code", "s");
        let live = reg.fresh("i");
        assert_eq!(
            live[0].recent_tool_use_ids.front().map(String::as_str),
            Some("toolu_keep")
        );
        assert_eq!(
            live[0].recent_prompt_hashes.front().copied(),
            Some(text_hash("keep me"))
        );
    }
}