openlatch-client 0.1.18

//! Human-identity signals for outbound hook envelopes (initiative I-1).
//!
//! Resolves the three optional CloudEvents extension attributes the platform
//! materialises as `agents.latest_os_user` / `latest_git_email` /
//! `latest_provider_account`, and hands them to the stamping gate in
//! `daemon/handlers.rs`. Nothing here decides anything: these are attribution
//! hints, never authorization input.
//!
//! # Where this module is, and is not
//!
//! Entirely daemon-side, behind `full-cli` by virtue of `src/lib.rs` gating the
//! whole `daemon` tree (D-04). `openlatch-hook` must not gain a byte of it — it
//! has a <3 ms / <20 MB budget and `ci/baselines/hook-deps.txt` is the
//! structural proof that this initiative cost it nothing.
//!
//! # Never block the event path (D-09)
//!
//! [`observe_session`] is synchronous and does one mutex lock. On a session's
//! first sight it marks the session `Pending`, spawns a one-shot task, and
//! returns *nothing to stamp*. The handler never awaits resolution.
//!
//! That means the first event of a session may carry no `gitemail` /
//! `provideracct`, and that is the accepted trade: the platform's `latest_*`
//! columns are latest-wins, so the next event of the same session fills them in.
//! Blocking even briefly would multiply across every session on the host, on the
//! one path an agent is waiting on.
//!
//! A plain `tokio::spawn` is correct here rather than
//! `core::supervision::spawn_supervised`: that harness restarts long-lived
//! subsystems under a backoff policy, and this task runs once, for well under a
//! second, with a failure mode — omit the signals — that is already the
//! module's normal answer. There is nothing for a supervisor to improve.
//!
//! # Fail closed
//!
//! Nothing in this module runs unless the resident policy bundle carries
//! `client_config.capture_identity_signals: true` (D-02). The gate lives at the
//! only call site, in `handlers.rs`.

pub mod git_email;
pub mod os_user;
pub mod provider_account;

use std::collections::VecDeque;
use std::sync::{LazyLock, Mutex};

pub use os_user::os_user;

/// Sessions held in the per-session memo.
///
/// A host runs a handful of concurrent agent sessions; 64 covers a busy
/// developer machine several times over, and a linear scan of 64 string
/// comparisons under a `Mutex` is cheaper than any map's hashing at this size.
/// Overflow is not a correctness problem — an evicted session re-resolves on its
/// next event.
const CACHE_CAP: usize = 64;

/// The two signals that need resolving off-thread. `osuser` is not among them:
/// it is process-global and already memoised in [`os_user`], so it is always
/// ready and never has to be omitted.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct IdentitySignals {
    pub git_email: Option<String>,
    pub provider_account: Option<String>,
}

/// Where one session sits in the resolution lifecycle.
///
/// `Pending` is inserted **synchronously, before the spawn**, and that ordering
/// is the duplicate-spawn guard: two events arriving back-to-back for a new
/// session cannot both see an empty cache and both spawn a resolver.
#[derive(Debug, Clone, PartialEq, Eq)]
enum SessionIdentity {
    Pending,
    Ready(IdentitySignals),
}

type IdentityCache = Mutex<VecDeque<(String, SessionIdentity)>>;

/// `(session_id, state)` in FIFO order.
///
/// The same shape as `PROJECT_ROOT_CACHE` in `config_monitor/enrich.rs` — a
/// second implementation of that idiom, not a reuse of it: there is no shared
/// helper, and the two caches hold different value types under different
/// lifetimes. Copied because it is proven and dependency-flat, not because it is
/// abstracted.
///
/// Deliberately **not** hung off `boundary::SessionRegistry`: that registry has
/// its own lifecycle and quiet-window eviction, and a core wire attribute should
/// not be a tenant in another subsystem's cache.
static CACHE: LazyLock<IdentityCache> =
    LazyLock::new(|| Mutex::new(VecDeque::with_capacity(CACHE_CAP)));

/// Counts how many resolutions actually ran. Test-only, and deliberately here
/// rather than inside the resolver stubs — plan 02 replaces those bodies, and
/// the duplicate-spawn guard must still be under test afterwards.
#[cfg(test)]
static RESOLVE_TASK_RUNS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

/// Serializes every identity test that touches process-global state.
///
/// All three resolvers read their inputs from the environment — the OS-user
/// ladder's `$SUDO_USER` / `$USER` / `$LOGNAME`, the git binary and the
/// `GIT_CONFIG*` family, the Claude state and settings files, the two
/// service-credential variables — and `cargo test` runs test functions on
/// parallel threads inside one process. Without this, one test's `set_var`
/// lands inside another's resolution: `git_email`'s `strip_git_env` walks
/// `vars_os()` while some other test is mutating it, and [`RESOLVE_TASK_RUNS`]
/// (a single global counter read as a delta) is shared by every test that
/// observes a session.
///
/// **Every** test in this module tree takes it, the synchronous ones included.
/// One that skipped it because its own variables looked private would
/// reintroduce exactly the race the lock exists to remove.
///
/// A `tokio::sync::Mutex` rather than a `std` one because the async tests hold
/// it across `await` points; the synchronous tests take it with
/// `blocking_lock`, which is safe there precisely because those functions run
/// outside a runtime.
#[cfg(test)]
pub(crate) static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

#[cfg(test)]
pub(crate) mod test_support {
    //! Fixtures shared by the tests across this module tree.
    //!
    //! Everything here exists because the three resolvers read process-global
    //! state, so their tests do too — and a fixture copied into four files is
    //! four places for the managed-variable list to drift apart.

    use std::ffi::{OsStr, OsString};
    use std::path::{Path, PathBuf};

    /// Every environment variable any identity test sets or clears.
    ///
    /// The union rather than a per-file subset: these tests share one process
    /// under one lock, so a guard that restored only the variables its own file
    /// touches would still hand the next test whatever a previous one left
    /// behind.
    const MANAGED: [&str; 14] = [
        // os_user
        "OPENLATCH_TEST_OS_USER",
        "SUDO_USER",
        "USER",
        "LOGNAME",
        // git_email
        "OPENLATCH_TEST_GIT_BIN",
        "GIT_CONFIG_COUNT",
        "GIT_CONFIG_KEY_0",
        "GIT_CONFIG_VALUE_0",
        "GIT_TEST_ASSUME_DIFFERENT_OWNER",
        // provider_account
        "OPENLATCH_TEST_CLAUDE_STATE_FILE",
        "OPENLATCH_TEST_CLAUDE_SETTINGS_FILE",
        "CLAUDE_CONFIG_DIR",
        "ANTHROPIC_API_KEY",
        "ANTHROPIC_AUTH_TOKEN",
    ];

    /// Saves every [`MANAGED`] variable, clears them all, and puts them back on
    /// drop.
    ///
    /// RAII rather than a `restore()` call at the end of the test body: a
    /// failing assertion unwinds straight past that call, and the variable it
    /// leaks then surfaces as an unrelated test failing elsewhere in the
    /// binary — the hardest kind of failure to read.
    ///
    /// Clearing on construction is what makes the cases mean anything: a
    /// developer with `ANTHROPIC_API_KEY` exported would otherwise resolve
    /// `shared-key` for every provider-account case in the matrix.
    pub(crate) struct EnvGuard {
        saved: Vec<(&'static str, Option<OsString>)>,
    }

    impl EnvGuard {
        pub(crate) fn clear() -> Self {
            let saved = MANAGED
                .iter()
                .map(|key| (*key, std::env::var_os(key)))
                .collect();
            for key in MANAGED {
                std::env::remove_var(key);
            }
            Self { saved }
        }

        pub(crate) fn set(&self, key: &str, value: impl AsRef<OsStr>) {
            std::env::set_var(key, value);
        }

        pub(crate) fn unset(&self, key: &str) {
            std::env::remove_var(key);
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            for (key, value) in &self.saved {
                match value {
                    Some(v) => std::env::set_var(key, v),
                    None => std::env::remove_var(key),
                }
            }
        }
    }

    /// Run a git command against `dir` and report whether it succeeded.
    pub(crate) fn git_ok(git: &Path, dir: &Path, args: &[&str]) -> bool {
        std::process::Command::new(git)
            .arg("-C")
            .arg(dir)
            .args(args)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .is_ok_and(|status| status.success())
    }

    /// Write `body` into `dir/name` and make it launchable.
    ///
    /// The body stays at the call site: the shims these tests need differ in
    /// what they do — sleep past the ceiling, or record every invocation before
    /// delegating — not in how they reach disk and become executable.
    pub(crate) fn write_shim(dir: &Path, name: &str, body: &str) -> PathBuf {
        let path = dir.join(name);
        std::fs::write(&path, body).expect("write shim");
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
                .expect("chmod shim");
        }
        path
    }
}

/// What to stamp on **this** event, and nothing more.
///
/// Returns the resolved signals when the session's resolution has finished, and
/// an empty set otherwise — whether that is because resolution is still in
/// flight or because this is the session's first event. The caller cannot tell
/// those apart, and does not need to: both mean "omit the attributes".
///
/// `cwd` is the working directory this event reported. It is only read on first
/// sight, because it is only the resolver's input; a session that later changes
/// directory keeps the email its repository gave it.
pub fn observe_session(session_id: &str, cwd: Option<&str>) -> IdentitySignals {
    let needs_resolution = {
        // A poisoned lock means some thread panicked mid-update. Degrade to
        // "stamp nothing" rather than propagate: identity is a hint, and taking
        // the hook path down over it would invert every priority this daemon has.
        let Ok(mut cache) = CACHE.lock() else {
            return IdentitySignals::default();
        };
        match cache.iter().position(|(id, _)| id == session_id) {
            Some(idx) => match &cache[idx].1 {
                SessionIdentity::Ready(signals) => return signals.clone(),
                SessionIdentity::Pending => false,
            },
            None => {
                insert_capped(&mut cache, session_id, SessionIdentity::Pending);
                true
            }
        }
    };

    if needs_resolution {
        tracing::debug!(
            target: "identity",
            session_id = %session_id,
            "resolving identity signals for a new session"
        );
        // The only two allocations on this path, and they happen once per
        // session rather than once per event — every later event for the same
        // session leaves here through the `Pending` arm above.
        tokio::spawn(resolve_task(session_id.to_owned(), cwd.map(str::to_owned)));
    }
    IdentitySignals::default()
}

/// Push an entry, evicting the oldest first when the ring is at its cap.
///
/// Shared by the two miss paths — a session seen for the first time and a
/// resolution landing after its session was evicted — because they must agree on
/// the eviction rule. Two copies would let the cap drift apart under editing.
fn insert_capped(
    cache: &mut VecDeque<(String, SessionIdentity)>,
    session_id: &str,
    state: SessionIdentity,
) {
    if cache.len() >= CACHE_CAP {
        cache.pop_front();
    }
    cache.push_back((session_id.to_owned(), state));
}

/// The one-shot resolution, run off the event path.
///
/// Both resolvers return `Option<String>` rather than `Result`: a missing
/// signal and a failed lookup are the same outcome — omit the attribute — and
/// neither is something the developer at the keyboard could act on, so neither
/// earns an `OL-` code (D-12).
///
/// The two lookups run **concurrently**. They share no input and no ordering —
/// one shells out to git in a working directory, the other reads a state file in
/// the home directory — so a sequential `await` pair would just add their
/// latencies together. Joining them here locks that in before plan 02 replaces
/// the stubs with the process spawns and file reads that make it matter.
async fn resolve_task(session_id: String, cwd: Option<String>) {
    #[cfg(test)]
    RESOLVE_TASK_RUNS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

    let (git_email, provider_account) = tokio::join!(
        git_email::resolve(cwd.as_deref()),
        provider_account::resolve()
    );
    store_ready(
        &session_id,
        IdentitySignals {
            git_email,
            provider_account,
        },
    );
}

/// Publish a finished resolution.
///
/// A session evicted while its resolution was in flight is **re-inserted**
/// rather than dropped: eviction says the ring is busy, not that this session
/// ended, and throwing the answer away would make the next event pay for the
/// same lookup again.
fn store_ready(session_id: &str, signals: IdentitySignals) {
    let Ok(mut cache) = CACHE.lock() else {
        return;
    };
    match cache.iter().position(|(id, _)| id == session_id) {
        Some(idx) => cache[idx].1 = SessionIdentity::Ready(signals),
        None => insert_capped(&mut cache, session_id, SessionIdentity::Ready(signals)),
    }
}

#[cfg(test)]
mod tests {
    use super::test_support::{git_ok, write_shim, EnvGuard};
    use super::*;
    use std::sync::atomic::Ordering;

    fn state_of(session_id: &str) -> Option<SessionIdentity> {
        let cache = CACHE.lock().expect("cache lock");
        cache
            .iter()
            .find(|(id, _)| id == session_id)
            .map(|(_, state)| state.clone())
    }

    /// Wait for a session's spawned resolver to publish its answer.
    ///
    /// The resolvers really do await now — two process spawns and a file read —
    /// so yielding alone is not enough on a busy runner; the loop sleeps
    /// between turns and gives up rather than hanging a stuck test forever.
    async fn settle(session_id: &str) {
        for _ in 0..600 {
            if matches!(state_of(session_id), Some(SessionIdentity::Ready(_))) {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
    }

    /// The whole cache contract in one sequential test.
    ///
    /// One function rather than five, because [`CACHE`] is process-global: split
    /// across `#[test]` functions the eviction case would evict the other cases'
    /// sessions out from under them on a parallel run.
    #[tokio::test]
    async fn cache_serves_pending_then_ready_and_evicts_fifo() {
        // [`RESOLVE_TASK_RUNS`] is one global counter read as a delta, so this
        // must not interleave with the memoisation test below.
        let _env = ENV_LOCK.lock().await;
        let runs_before = RESOLVE_TASK_RUNS.load(Ordering::Relaxed);
        let first_session = "sess_identity_first";

        // First sight: nothing to stamp, and the session is marked before the
        // spawn so a racing second event cannot spawn a second resolver.
        assert_eq!(
            observe_session(first_session, Some("/repo")),
            IdentitySignals::default()
        );
        assert_eq!(state_of(first_session), Some(SessionIdentity::Pending));

        // Second event while still resolving: still nothing to stamp, still one
        // resolver.
        assert_eq!(
            observe_session(first_session, Some("/repo")),
            IdentitySignals::default()
        );

        settle(first_session).await;
        assert_eq!(
            RESOLVE_TASK_RUNS.load(Ordering::Relaxed) - runs_before,
            1,
            "one resolution per session, however many events arrive"
        );
        assert!(
            matches!(state_of(first_session), Some(SessionIdentity::Ready(_))),
            "the finished resolution is published back into the cache"
        );

        // A Ready session is served from the cache and spawns nothing. Seeded
        // with addresses no resolver on this host could produce, so the
        // assertion proves the cached entry was served back verbatim rather
        // than quietly re-resolved.
        let ready_session = "sess_identity_ready";
        let resolved = IdentitySignals {
            git_email: Some("dev@example.com".to_string()),
            provider_account: Some("dev@anthropic.example".to_string()),
        };
        store_ready(ready_session, resolved.clone());
        let runs_before_ready = RESOLVE_TASK_RUNS.load(Ordering::Relaxed);
        assert_eq!(observe_session(ready_session, None), resolved);
        assert_eq!(
            RESOLVE_TASK_RUNS.load(Ordering::Relaxed),
            runs_before_ready,
            "a cache hit resolves nothing"
        );

        // FIFO eviction at the cap. Asserted before yielding, so the resolvers
        // these calls spawned cannot re-insert anything mid-assertion.
        for n in 0..CACHE_CAP {
            observe_session(&format!("sess_identity_filler_{n}"), None);
        }
        assert_eq!(
            state_of(first_session),
            None,
            "the oldest session is evicted once the ring is full"
        );
        assert_eq!(
            CACHE.lock().expect("cache lock").len(),
            CACHE_CAP,
            "the ring never grows past its cap"
        );
    }

    // -- A10: resolution is once per session, not once per event -------------

    /// A git that records every invocation before handing off to the real one.
    ///
    /// The recording is what makes the claim checkable: the resolver's own
    /// bookkeeping could say "one resolution" while still spawning a process
    /// per event, and it is the spawns that a developer's machine feels.
    fn counting_shim(
        dir: &std::path::Path,
        real_git: &std::path::Path,
        log: &std::path::Path,
    ) -> std::path::PathBuf {
        let (name, body) = if cfg!(windows) {
            (
                "counting-git.cmd",
                format!(
                    "@echo off\r\n>>\"{log}\" echo %*\r\n\"{git}\" %*\r\n",
                    log = log.display(),
                    git = real_git.display()
                ),
            )
        } else {
            (
                "counting-git.sh",
                format!(
                    "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"{log}\"\nexec \"{git}\" \"$@\"\n",
                    log = log.display(),
                    git = real_git.display()
                ),
            )
        };
        write_shim(dir, name, &body)
    }

    /// Invocations the shim recorded for one working directory.
    ///
    /// Filtered by directory rather than counting every line, so a sibling
    /// test resolving some other session cannot inflate the number.
    fn git_calls(log: &std::path::Path, cwd: &str) -> usize {
        std::fs::read_to_string(log)
            .unwrap_or_default()
            .lines()
            .filter(|line| line.contains(cwd))
            .count()
    }

    /// A10 — events for one session cost one resolution, however many arrive.
    ///
    /// The number that matters is the process count, not the task count: a
    /// resolver that ran per event would be invisible to
    /// [`RESOLVE_TASK_RUNS`]-style bookkeeping if the bookkeeping moved, but a
    /// counting git records every spawn regardless of how the code is shaped.
    ///
    /// The assertions are all relative — "unchanged", "the same again" — never
    /// a literal spawn count. The claim under test is that the cost does not
    /// grow with events; how many spawns one resolution takes is the resolver's
    /// business, and pinning it here would turn every change there into a
    /// failure in this file.
    #[tokio::test]
    async fn identity_memoisation_once_per_session() {
        let _env = ENV_LOCK.lock().await;
        let Some(real_git) = git_email::discover_platform() else {
            return;
        };

        let workspace = tempfile::tempdir().expect("tempdir");
        let repo = workspace.path().join("repo");
        std::fs::create_dir(&repo).expect("repo dir");
        if !git_ok(&real_git, &repo, &["init", "-q"]) {
            return;
        }
        assert!(
            git_ok(
                &real_git,
                &repo,
                &["config", "user.email", "memo@fixture.test"]
            ),
            "seed the repo-local address"
        );

        let log = workspace.path().join("git-invocations.log");
        let shim = counting_shim(workspace.path(), &real_git, &log);
        let state = workspace.path().join("state.json");
        std::fs::write(
            &state,
            r#"{"oauthAccount":{"emailAddress":"alice@fixture.test"}}"#,
        )
        .expect("write state fixture");

        // Both resolvers are driven entirely from the environment, and the
        // guard clears the whole managed set — including the two
        // service-credential variables, which a developer with one exported
        // would otherwise see resolve to `shared-key` instead of the address.
        let env = EnvGuard::clear();
        env.set("OPENLATCH_TEST_GIT_BIN", &shim);
        env.set("OPENLATCH_TEST_CLAUDE_STATE_FILE", &state);

        let cwd = repo.to_str().expect("utf-8 path");
        let runs_before = RESOLVE_TASK_RUNS.load(Ordering::Relaxed);

        let first = "sess_identity_memo_first";
        for _ in 0..5 {
            observe_session(first, Some(cwd));
        }
        settle(first).await;
        let per_session = git_calls(&log, cwd);
        assert!(
            per_session > 0,
            "the resolution really did reach git — otherwise nothing below is measuring anything"
        );

        // The sixth call is served entirely from the cache, and carries the
        // values the two resolvers actually produced.
        assert_eq!(
            observe_session(first, Some(cwd)),
            IdentitySignals {
                git_email: Some("memo@fixture.test".to_string()),
                provider_account: Some("alice@fixture.test".to_string()),
            }
        );
        assert_eq!(
            RESOLVE_TASK_RUNS.load(Ordering::Relaxed) - runs_before,
            1,
            "six events, one resolution"
        );

        // More events for a session that has already resolved. The sleep gives
        // a resolver that this code wrongly spawned time to reach the log, so
        // "unchanged" is a real observation rather than a race won.
        for _ in 0..5 {
            observe_session(first, Some(cwd));
        }
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        assert_eq!(
            git_calls(&log, cwd),
            per_session,
            "a resolved session costs nothing further, however many events arrive"
        );

        // A different session is a different working directory as far as this
        // module knows, so it resolves again — and costs exactly what the first
        // one cost, not one spawn more.
        let second = "sess_identity_memo_second";
        observe_session(second, Some(cwd));
        settle(second).await;
        assert_eq!(
            RESOLVE_TASK_RUNS.load(Ordering::Relaxed) - runs_before,
            2,
            "a new session resolves once more"
        );
        assert_eq!(
            git_calls(&log, cwd) - per_session,
            per_session,
            "the second session pays the same fixed per-session cost as the first"
        );
    }
}