swapdex 0.167.16

//! Attribute sessions to the account that was active when they ran, by joining
//! the switch `timeline` with session start times. Attribution is best-effort:
//! a session with no prior switch event is `unattributed` (a first-class
//! bucket), and a missing/older sessionwiki degrades gracefully (A14).

use crate::paths::Paths;
use serde_json::Value;
use std::collections::BTreeMap;

pub const UNATTRIBUTED: &str = "(unattributed)";

/// The ledger's name for an account that has been removed. Parenthesised like
/// `UNATTRIBUTED` so it reads as a bucket rather than an account, and so no
/// account that later takes the freed name can be matched against it.
pub fn retired(name: &str) -> String {
    format!("(removed: {name})")
}

pub struct Event {
    pub ts: i64,
    pub tool: String,
    pub account: String,
    /// What was done: `use` / `restore` move where sessions live, `serve` moves
    /// who pays for them. The reader used to drop this, so every event answered
    /// both questions - and once `serve` started writing here, a change of payer
    /// would have been read as a change of where the conversation lives.
    pub action: String,
}

/// Actions that move where new sessions start (and so which account holds a
/// conversation). An entry written before actions were read carries none, and
/// those were all switches, so a missing action counts as one.
fn is_switch(action: &str) -> bool {
    matches!(action, "" | "use" | "restore")
}

/// The action `serve` writes: turns handed to an account without moving the
/// conversations.
pub const SERVE: &str = "serve";

/// Explicit passthrough: no managed account can be named as payer from here.
pub const SERVE_OFF: &str = "serve-off";

fn changes_payer(action: &str) -> bool {
    is_switch(action) || matches!(action, SERVE | SERVE_OFF)
}

fn latest_event<'a>(
    events: &'a [Event],
    tool: &str,
    at_secs: i64,
    mut include: impl FnMut(&Event) -> bool,
) -> Option<&'a Event> {
    events
        .iter()
        .enumerate()
        .filter(|(_, event)| event.tool == tool && event.ts <= at_secs && include(event))
        .max_by_key(|(order, event)| (event.ts, *order))
        .map(|(_, event)| event)
}

pub fn read_timeline(paths: &Paths) -> Vec<Event> {
    let path = paths.store_dir().join("timeline.jsonl");
    let mut out = Vec::new();
    if let Ok(text) = std::fs::read_to_string(path) {
        for line in text.lines() {
            if let Ok(v) = serde_json::from_str::<Value>(line) {
                if let (Some(ts), Some(tool), Some(account)) =
                    (v["ts"].as_i64(), v["tool"].as_str(), v["account"].as_str())
                {
                    out.push(Event {
                        ts,
                        tool: tool.to_string(),
                        account: account.to_string(),
                        action: v["action"].as_str().unwrap_or_default().to_string(),
                    });
                }
            }
        }
    }
    out
}

/// The account active when a session of `tool` started: the last switch event
/// for that tool with `ts <= started`. None (unattributed) if none precedes it.
pub fn attribute(events: &[Event], tool: &str, started_secs: i64) -> Option<String> {
    latest_event(events, tool, started_secs, |event| is_switch(&event.action))
        .map(|event| event.account.clone())
}

/// The account `tool` was ON at `at_secs`: the newest event of ANY action at or
/// before it, `None` before the first one.
///
/// `attribute` answers a narrower question - where a `use` last moved the
/// conversations - and it was standing in for this one. That held only while
/// every switch went through `use`. Once switching moved into the proxy the
/// last `use` on this machine went months stale and every change since was a
/// `serve`, so sessions were credited to an account that had not served a turn
/// in weeks; a tool never `use`d at all was credited to nobody, which was 4457
/// of 5193 sessions here.
pub fn active_at(events: &[Event], tool: &str, at_secs: i64) -> Option<String> {
    latest_event(events, tool, at_secs, |event| changes_payer(&event.action)).and_then(|event| {
        (event.action != SERVE_OFF && !event.account.is_empty()).then(|| event.account.clone())
    })
}

/// The account PAYING for `tool` at `at_secs`: the newest payer-changing event,
/// ordered by timestamp and append position. Explicit passthrough has no known
/// managed payer.
///
/// A Codex transcript's rate limits come from the token that served those turns,
/// so this is the account they describe. Reading them off the home the file sits
/// in reports one account's usage under another's name.
pub fn payer_at(events: &[Event], tool: &str, at_secs: i64) -> Option<String> {
    latest_event(events, tool, at_secs, |event| changes_payer(&event.action)).and_then(|event| {
        (event.action != SERVE_OFF && !event.account.is_empty()).then(|| event.account.clone())
    })
}

/// Session counts per account, best-effort from `sessionwiki list --json`. None
/// if sessionwiki is absent/unusable - the caller degrades to "unavailable".
pub fn sessions_by_account(paths: &Paths) -> Option<BTreeMap<String, usize>> {
    let rows = sessionwiki_rows()?;
    Some(count_by_account(&rows, &read_timeline(paths)))
}

/// Pure counting, separated from the sessionwiki shell-out so which account a
/// row lands under is unit-testable.
pub(crate) fn count_by_account(rows: &[Value], events: &[Event]) -> BTreeMap<String, usize> {
    let mut counts: BTreeMap<String, usize> = BTreeMap::new();
    for row in rows {
        let tool = match row["tool"].as_str() {
            Some(t) => t,
            None => continue,
        };
        let started = row["started"]
            .as_str()
            .and_then(rfc3339_to_secs)
            .unwrap_or(0);
        let acct = active_at(events, tool, started).unwrap_or_else(|| UNATTRIBUTED.to_string());
        *counts.entry(acct).or_insert(0) += 1;
    }
    counts
}

pub fn status_line(paths: &Paths) -> Option<String> {
    // Until at least one switch is recorded, every session is unattributed and
    // the count is just the fetch cap - a confusing "N across 0 account(s)".
    // Say nothing rather than mislead.
    if read_timeline(paths).is_empty() {
        return None;
    }
    let counts = sessions_by_account(paths)?;
    let total: usize = counts.values().sum();
    if total == 0 {
        // Fresh install: sessionwiki is present but never synced. Point at
        // the cure instead of claiming "0 sessions" on a full disk.
        return Some("sessions: index empty - run `sessionwiki sync` once".into());
    }
    let unattributed = counts.get(UNATTRIBUTED).copied().unwrap_or(0);
    let accounts = counts.keys().filter(|k| *k != UNATTRIBUTED).count();
    let tail = if unattributed > 0 {
        format!(", {unattributed} unattributed")
    } else {
        String::new()
    };
    Some(format!(
        "sessions: {total} across {accounts} account(s){tail} (sessionwiki)"
    ))
}

/// A session row for the post-switch continuity hint in `ui`.
pub struct RecentSession {
    pub id: String,
    pub tool: String,
    pub title: String,
    pub started: i64,
}

/// The most recent sessions attributed to `account`, newest first. None when
/// sessionwiki is absent (the caller simply shows no hint).
pub fn recent_sessions_for(paths: &Paths, account: &str, n: usize) -> Option<Vec<RecentSession>> {
    recent_sessions_for_tool(paths, account, n, None)
}

pub(crate) fn recent_sessions_for_tool(
    paths: &Paths,
    account: &str,
    n: usize,
    tool: Option<&str>,
) -> Option<Vec<RecentSession>> {
    let rows = sessionwiki_rows_for_tool(tool)?;
    let events = read_timeline(paths);
    Some(pick_recent(&rows, &events, account, n, tool))
}

/// The most recent sessions regardless of account - the honest fallback for a
/// store with no switch history yet (nothing can be attributed before the
/// first switch). None when sessionwiki is absent.
pub fn recent_sessions_any(n: usize) -> Option<Vec<RecentSession>> {
    recent_sessions_any_for_tool(n, None)
}

pub(crate) fn recent_sessions_any_for_tool(
    n: usize,
    tool: Option<&str>,
) -> Option<Vec<RecentSession>> {
    let rows = sessionwiki_rows_for_tool(tool)?;
    let mut out: Vec<RecentSession> = rows
        .iter()
        .filter_map(|row| {
            Some(RecentSession {
                id: row["id"].as_str()?.to_string(),
                tool: row["tool"].as_str()?.to_string(),
                title: row["title"].as_str().unwrap_or("(untitled)").to_string(),
                started: row["started"].as_str().and_then(rfc3339_to_secs)?,
            })
        })
        .collect();
    out.sort_by_key(|s| std::cmp::Reverse(s.started));
    out.truncate(n);
    Some(out)
}

/// Pure selection: filter rows to those attributed to `account`, newest first,
/// top `n`. Separated from the sessionwiki shell-out so it is unit-testable.
pub(crate) fn pick_recent(
    rows: &[Value],
    events: &[Event],
    account: &str,
    n: usize,
    selected_tool: Option<&str>,
) -> Vec<RecentSession> {
    let mut out: Vec<RecentSession> = rows
        .iter()
        .filter_map(|row| {
            let tool = row["tool"].as_str()?;
            if selected_tool.is_some_and(|selected| selected != tool) {
                return None;
            }
            let started = row["started"].as_str().and_then(rfc3339_to_secs)?;
            if active_at(events, tool, started).as_deref() != Some(account) {
                return None;
            }
            Some(RecentSession {
                id: row["id"].as_str()?.to_string(),
                tool: tool.to_string(),
                title: row["title"].as_str().unwrap_or("(untitled)").to_string(),
                started,
            })
        })
        .collect();
    out.sort_by_key(|s| std::cmp::Reverse(s.started));
    out.truncate(n);
    out
}

/// Run `sessionwiki list --json --no-sync` bounded by a short timeout, parsing
/// defensively. Any failure (absent binary, non-zero exit, unparseable, slow)
/// returns None so `status`/`sessions` never hangs or errors.
fn sessionwiki_rows() -> Option<Vec<Value>> {
    sessionwiki_rows_for_tool(None)
}

fn terminate_process_group(child: &mut std::process::Child) {
    let Ok(group) = i32::try_from(child.id()) else {
        let _ = child.kill();
        let _ = child.wait();
        return;
    };
    unsafe {
        libc::kill(-group, libc::SIGTERM);
    }
    let grace = std::time::Instant::now() + std::time::Duration::from_millis(200);
    while std::time::Instant::now() < grace {
        if child.try_wait().ok().flatten().is_some() {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(10));
    }
    // The leader can exit before a descendant. Address the group once more,
    // then always wait for the direct child so it cannot become a zombie.
    unsafe {
        libc::kill(-group, libc::SIGKILL);
    }
    let _ = child.wait();
}

fn run_sessionwiki(
    program: &std::ffi::OsStr,
    tool: Option<&str>,
    timeout: std::time::Duration,
) -> Option<std::process::Output> {
    use std::io::{Read, Seek};
    use std::os::unix::process::CommandExt;
    use std::process::{Command, Stdio};

    // Files keep a verbose sessionwiki response from filling an unread pipe
    // while the parent is polling the process deadline.
    let mut stdout = tempfile::tempfile().ok()?;
    let mut stderr = tempfile::tempfile().ok()?;
    let mut cmd = Command::new(program);
    cmd.args(["list", "--json", "--no-sync", "-n", "50000"]);
    if let Some(tool) = tool {
        cmd.args(["--tool", tool]);
    }
    cmd.stdin(Stdio::null())
        .stdout(Stdio::from(stdout.try_clone().ok()?))
        .stderr(Stdio::from(stderr.try_clone().ok()?))
        .process_group(0);
    let mut child = cmd.spawn().ok()?;
    let deadline = std::time::Instant::now() + timeout;
    let status = loop {
        match child.try_wait() {
            Ok(Some(status)) => break status,
            Ok(None) if std::time::Instant::now() < deadline => {
                std::thread::sleep(std::time::Duration::from_millis(10));
            }
            Ok(None) | Err(_) => {
                terminate_process_group(&mut child);
                return None;
            }
        }
    };
    stdout.rewind().ok()?;
    stderr.rewind().ok()?;
    let mut stdout_bytes = Vec::new();
    let mut stderr_bytes = Vec::new();
    stdout.read_to_end(&mut stdout_bytes).ok()?;
    stderr.read_to_end(&mut stderr_bytes).ok()?;
    Some(std::process::Output {
        status,
        stdout: stdout_bytes,
        stderr: stderr_bytes,
    })
}

fn sessionwiki_rows_for_tool(tool: Option<&str>) -> Option<Vec<Value>> {
    let matches_tool = |row: &&Value| tool.is_none_or(|t| row["tool"].as_str() == Some(t));
    // Test hook: a fixture file stands in for the shell-out so the ui flow is
    // E2E-testable inside an isolated root. Only honored WITH SWAPDEX_ROOT so
    // a stray env var can never redirect a production run.
    if let Some(p) = std::env::var_os("SWAPDEX_SESSIONWIKI_JSON")
        .filter(|_| std::env::var_os("SWAPDEX_ROOT").is_some())
    {
        let v: Value = serde_json::from_slice(&std::fs::read(p).ok()?).ok()?;
        return Some(v.as_array()?.iter().filter(matches_tool).cloned().collect());
    }
    // Under a dev/test root, sessionwiki would still read the HOST's real
    // sessions (it has no notion of SWAPDEX_ROOT), leaking them into an isolated
    // run. Skip it entirely in that mode.
    if std::env::var_os("SWAPDEX_ROOT").is_some() {
        return None;
    }
    let out = run_sessionwiki(
        std::ffi::OsStr::new("sessionwiki"),
        tool,
        std::time::Duration::from_secs(5),
    )?;
    if !out.status.success() {
        return None;
    }
    let v: Value = serde_json::from_slice(&out.stdout).ok()?;
    Some(v.as_array()?.iter().filter(matches_tool).cloned().collect())
}

pub fn rfc3339_to_secs(s: &str) -> Option<i64> {
    // Minimal parse: "YYYY-MM-DDTHH:MM:SS...": compute epoch seconds. Avoid a
    // chrono dep; only the ordering vs timeline ts matters, so a coarse but
    // monotonic value is fine. Fall back on any deviation.
    let bytes = s.as_bytes();
    if s.len() < 19 || bytes.get(4) != Some(&b'-') {
        return None;
    }
    let g = |a: usize, b: usize| s.get(a..b)?.parse::<i64>().ok();
    let (y, mo, d) = (g(0, 4)?, g(5, 7)?, g(8, 10)?);
    let (h, mi, se) = (g(11, 13)?, g(14, 16)?, g(17, 19)?);
    // days since epoch via a civil-from-date algorithm (Howard Hinnant).
    let yy = if mo <= 2 { y - 1 } else { y };
    let era = yy.div_euclid(400);
    let yoe = yy - era * 400;
    let doy = (153 * (if mo > 2 { mo - 3 } else { mo + 9 }) + 2) / 5 + d - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    let days = era * 146097 + doe - 719468;
    let naive = days * 86400 + h * 3600 + mi * 60 + se;
    // Normalize a trailing +HH:MM / -HH:MM offset to true UTC (UTC = local -
    // offset). A trailing "Z" or nothing is already UTC. Skip past fractional
    // seconds first ("...:00.123+09:00") or the offset would be missed.
    let mut off = &s[19..];
    if let Some(rest) = off.strip_prefix('.') {
        let digits = rest.chars().take_while(|c| c.is_ascii_digit()).count();
        off = &rest[digits..];
    }
    let offset_secs = if let Some(rest) = off.strip_prefix('+').or_else(|| off.strip_prefix('-')) {
        let sign = if off.starts_with('-') { -1 } else { 1 };
        let oh: i64 = rest.get(0..2).and_then(|x| x.parse().ok()).unwrap_or(0);
        let om: i64 = rest.get(3..5).and_then(|x| x.parse().ok()).unwrap_or(0);
        sign * (oh * 3600 + om * 60)
    } else {
        0
    };
    Some(naive - offset_secs)
}

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

    #[cfg(unix)]
    fn process_exists(pid: i32) -> bool {
        let result = unsafe { libc::kill(pid, 0) };
        result == 0 || std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH)
    }

    #[cfg(unix)]
    #[test]
    fn slow_sessionwiki_is_terminated_with_its_descendants() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let pid_file = dir.path().join("pids");
        let fake = dir.path().join("sessionwiki");
        std::fs::write(
            &fake,
            format!(
                "#!/bin/sh\n\
                 sleep 30 &\n\
                 child=$!\n\
                 printf '%s %s\\n' \"$$\" \"$child\" > '{}'\n\
                 trap 'kill \"$child\" 2>/dev/null; wait \"$child\" 2>/dev/null; exit 143' TERM INT\n\
                 wait \"$child\"\n",
                pid_file.display()
            ),
        )
        .unwrap();
        let mut permissions = std::fs::metadata(&fake).unwrap().permissions();
        permissions.set_mode(0o755);
        std::fs::set_permissions(&fake, permissions).unwrap();

        // A run that never reached the fake's first line - exec refused with
        // ETXTBSY while a sibling test forked, or a loaded runner too slow to
        // start it within the timeout - says nothing about termination, so
        // only those runs are tried again.
        for _ in 0..5 {
            let started = std::time::Instant::now();
            let output = run_sessionwiki(
                fake.as_os_str(),
                None,
                std::time::Duration::from_millis(250),
            );
            assert!(output.is_none(), "a timed-out command has no usable output");
            assert!(
                started.elapsed() < std::time::Duration::from_secs(2),
                "the timeout is bounded"
            );
            if pid_file.exists() {
                break;
            }
        }

        let pids: Vec<i32> = std::fs::read_to_string(&pid_file)
            .expect("the fake started and recorded both processes")
            .split_whitespace()
            .map(|pid| pid.parse().unwrap())
            .collect();
        assert_eq!(pids.len(), 2);
        for _ in 0..50 {
            if pids.iter().all(|pid| !process_exists(*pid)) {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        let survivors: Vec<i32> = pids
            .iter()
            .copied()
            .filter(|pid| process_exists(*pid))
            .collect();
        for pid in &survivors {
            unsafe {
                libc::kill(*pid, libc::SIGKILL);
            }
        }
        assert!(survivors.is_empty(), "processes still alive: {survivors:?}");
    }

    #[cfg(unix)]
    #[test]
    fn completed_sessionwiki_output_is_captured() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let fake = dir.path().join("sessionwiki");
        std::fs::write(&fake, "#!/bin/sh\nprintf '[]\\n'\n").unwrap();
        let mut permissions = std::fs::metadata(&fake).unwrap().permissions();
        permissions.set_mode(0o755);
        std::fs::set_permissions(&fake, permissions).unwrap();

        let output = run_sessionwiki(
            fake.as_os_str(),
            Some("codex"),
            std::time::Duration::from_secs(1),
        )
        .expect("fast successful command");
        assert!(output.status.success());
        assert_eq!(output.stdout, b"[]\n");
    }

    fn ev(ts: i64, tool: &str, acct: &str) -> Event {
        Event {
            ts,
            tool: tool.into(),
            account: acct.into(),
            action: "use".into(),
        }
    }

    /// A `serve` event: who pays, not where the conversation lives.
    fn served(ts: i64, tool: &str, acct: &str) -> Event {
        Event {
            action: SERVE.into(),
            ..ev(ts, tool, acct)
        }
    }

    #[test]
    fn a_serve_event_never_moves_where_a_session_is_attributed() {
        let events = vec![ev(100, "codex", "home"), served(200, "codex", "payer")];
        assert_eq!(
            attribute(&events, "codex", 300).as_deref(),
            Some("home"),
            "the conversation stayed where `use` put it"
        );
        assert_eq!(
            payer_at(&events, "codex", 300).as_deref(),
            Some("payer"),
            "and the turns were paid for elsewhere"
        );
        assert_eq!(
            payer_at(&events, "codex", 150).as_deref(),
            Some("home"),
            "before anyone was handed the turns, the home account paid"
        );
    }

    fn restored(ts: i64, tool: &str, acct: &str) -> Event {
        Event {
            action: "restore".into(),
            ..ev(ts, tool, acct)
        }
    }

    fn serving_off(ts: i64, tool: &str) -> Event {
        Event {
            ts,
            tool: tool.into(),
            account: String::new(),
            action: "serve-off".into(),
        }
    }

    /// Every action that changes who pays competes in one chronological stream.
    /// An older `serve` must not outrank a newer `use` or `restore`, and explicit
    /// off means the payer is unknown until another state-changing event.
    #[test]
    fn payer_follows_the_latest_state_change_and_off_is_unknown() {
        let mut events = vec![
            ev(100, "codex", "home"),
            served(200, "codex", "payer"),
            ev(300, "codex", "work"),
        ];
        assert_eq!(payer_at(&events, "codex", 250).as_deref(), Some("payer"));
        assert_eq!(payer_at(&events, "codex", 350).as_deref(), Some("work"));

        events.push(restored(400, "codex", "home"));
        assert_eq!(payer_at(&events, "codex", 450).as_deref(), Some("home"));

        events.push(serving_off(500, "codex"));
        assert_eq!(
            payer_at(&events, "codex", 550),
            None,
            "passthrough tokens have no managed payer to fall back to"
        );
        assert_eq!(
            attribute(&events, "codex", 550).as_deref(),
            Some("home"),
            "off changes payment, not where sessions launch"
        );

        events.push(served(600, "codex", "payer"));
        assert_eq!(payer_at(&events, "codex", 650).as_deref(), Some("payer"));
    }

    /// Timeline seconds are deliberately coarse. When several commands land in
    /// one second, append order is the only ordering left and the last appended
    /// state must win, without leaking across tools.
    #[test]
    fn equal_timestamp_payer_events_use_append_order_per_tool() {
        let mut events = vec![
            served(100, "codex", "first"),
            ev(100, "codex", "second"),
            served(100, "claude-code", "claude-payer"),
        ];
        assert_eq!(payer_at(&events, "codex", 100).as_deref(), Some("second"));
        assert_eq!(
            payer_at(&events, "claude-code", 100).as_deref(),
            Some("claude-payer")
        );

        events.push(serving_off(100, "codex"));
        assert_eq!(payer_at(&events, "codex", 100), None);
        assert_eq!(
            payer_at(&events, "claude-code", 100).as_deref(),
            Some("claude-payer"),
            "Codex off does not change Claude"
        );

        events.push(served(100, "codex", "reenabled"));
        assert_eq!(
            payer_at(&events, "codex", 100).as_deref(),
            Some("reenabled")
        );
    }

    #[test]
    fn the_account_a_session_ran_under_is_the_newest_evidence() {
        // A machine driven by the proxy: the last `use` is months old and every
        // change of account since has been a `serve`.
        let events = vec![ev(100, "codex", "stale"), served(200, "codex", "current")];
        assert_eq!(
            active_at(&events, "codex", 300).as_deref(),
            Some("current"),
            "the newest evidence wins, whatever wrote it"
        );
        assert_eq!(
            active_at(&events, "codex", 50),
            None,
            "still nothing before the first event"
        );

        // And a tool that has never been `use`d at all on this machine.
        let only_serves = vec![served(200, "claude-code", "kong")];
        assert_eq!(
            active_at(&only_serves, "claude-code", 300).as_deref(),
            Some("kong")
        );
        assert_eq!(
            attribute(&only_serves, "claude-code", 300),
            None,
            "which the switch-only reader could not see"
        );
    }

    #[test]
    fn attribute_picks_the_last_switch_before_the_session() {
        let events = vec![
            ev(100, "codex", "work"),
            ev(200, "codex", "home"),
            ev(150, "claude-code", "personal"),
        ];
        // started at 250 -> last codex switch was 200 (home)
        assert_eq!(attribute(&events, "codex", 250).as_deref(), Some("home"));
        // started at 150 -> codex switch at 100 (work), not the 200 (later)
        assert_eq!(attribute(&events, "codex", 150).as_deref(), Some("work"));
        // started before ANY codex switch -> unattributed
        assert_eq!(attribute(&events, "codex", 50), None);
        // different tool timeline is independent
        assert_eq!(
            attribute(&events, "claude-code", 160).as_deref(),
            Some("personal")
        );
    }

    #[test]
    fn rfc3339_orders_correctly() {
        let a = rfc3339_to_secs("2026-06-10T10:00:00+00:00").unwrap();
        let b = rfc3339_to_secs("2026-06-10T10:00:01+00:00").unwrap();
        assert_eq!(b - a, 1);
        assert!(rfc3339_to_secs("2027-01-01T00:00:00Z").unwrap() > a);
    }

    #[test]
    fn rfc3339_offset_applies_after_fractional_seconds() {
        // A +09:00 offset behind fractional seconds must still normalize to
        // UTC (it used to be silently ignored).
        let utc = rfc3339_to_secs("2026-06-10T01:00:00Z").unwrap();
        let kst = rfc3339_to_secs("2026-06-10T10:00:00.123+09:00").unwrap();
        assert_eq!(kst, utc);
    }

    #[test]
    fn pick_recent_filters_by_account_and_orders_newest_first() {
        // Switch timeline: work until t=100, then personal from t=100.
        let events = vec![ev(50, "codex", "work"), ev(100, "codex", "personal")];
        let rows: Vec<serde_json::Value> = vec![
            serde_json::json!({"id":"aaa111","tool":"codex","title":"on work",
                               "started":"1970-01-01T00:01:00Z"}), // t=60 -> work
            serde_json::json!({"id":"bbb222","tool":"codex","title":"newer on personal",
                               "started":"1970-01-01T00:03:00Z"}), // t=180 -> personal
            serde_json::json!({"id":"ccc333","tool":"codex","title":"older on personal",
                               "started":"1970-01-01T00:02:00Z"}), // t=120 -> personal
        ];
        let got = pick_recent(&rows, &events, "personal", 5, None);
        let ids: Vec<&str> = got.iter().map(|s| s.id.as_str()).collect();
        assert_eq!(ids, vec!["bbb222", "ccc333"], "personal only, newest first");
        let one = pick_recent(&rows, &events, "personal", 1, None);
        assert_eq!(one.len(), 1, "truncates to n");
        assert_eq!(one[0].id, "bbb222");
    }

    #[test]
    fn provider_selection_precedes_the_recent_session_limit() {
        let events = vec![ev(0, "codex", "shared"), ev(0, "claude-code", "shared")];
        let rows = vec![
            serde_json::json!({"id":"claude-new", "tool":"claude-code", "started":"1970-01-01T00:03:00Z"}),
            serde_json::json!({"id":"codex-old", "tool":"codex", "started":"1970-01-01T00:02:00Z"}),
        ];
        let got = pick_recent(&rows, &events, "shared", 1, Some("codex"));
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].id, "codex-old");
    }

    #[test]
    fn a_served_session_is_not_counted_as_unattributed() {
        // claude-code on this machine has never been `use`d, so the whole
        // breakdown read "(unattributed)" while every turn had a named payer.
        let events = vec![
            served(50, "claude-code", "rnd"),
            served(100, "claude-code", "kong"),
        ];
        let rows: Vec<serde_json::Value> = vec![
            serde_json::json!({"tool":"claude-code","started":"1970-01-01T00:01:00Z"}), // t=60
            serde_json::json!({"tool":"claude-code","started":"1970-01-01T00:03:00Z"}), // t=180
            serde_json::json!({"tool":"claude-code","started":"1970-01-01T00:00:10Z"}), // t=10
        ];
        let counts = count_by_account(&rows, &events);
        assert_eq!(counts.get("rnd"), Some(&1));
        assert_eq!(counts.get("kong"), Some(&1));
        assert_eq!(
            counts.get(UNATTRIBUTED),
            Some(&1),
            "only the session that predates every event stays unattributed"
        );
    }

    #[test]
    fn pick_recent_sees_sessions_on_a_proxy_driven_machine() {
        // The shape that produced 4457 unattributed sessions: the account
        // changed through the proxy, so the only events are serves.
        let events = vec![
            served(50, "claude-code", "rnd"),
            served(100, "claude-code", "kong"),
        ];
        let rows: Vec<serde_json::Value> = vec![
            serde_json::json!({"id":"aaa111","tool":"claude-code","title":"on rnd",
                               "started":"1970-01-01T00:01:00Z"}), // t=60 -> rnd
            serde_json::json!({"id":"bbb222","tool":"claude-code","title":"on kong",
                               "started":"1970-01-01T00:03:00Z"}), // t=180 -> kong
        ];
        let ids: Vec<String> = pick_recent(&rows, &events, "kong", 5, None)
            .iter()
            .map(|s| s.id.clone())
            .collect();
        assert_eq!(
            ids,
            vec!["bbb222"],
            "a served session belongs to the account that served it"
        );
        assert_eq!(
            pick_recent(&rows, &events, "rnd", 5, None).len(),
            1,
            "and the earlier one to the account serving then"
        );
    }
}