diff --git a/crates/statsai-adapters/src/claude/auth/desktop.rs b/crates/statsai-adapters/src/claude/auth/desktop.rs new file mode 100644 index 0000000..40dcf7f --- /dev/null +++ b/crates/statsai-adapters/src/claude/auth/desktop.rs @@ -0,0 +1,594 @@ +//! Per-conversation account evidence from the Claude desktop app. +//! +//! The desktop app runs Claude Code under its own sign-in and writes the same +//! `~/.claude/projects` transcripts as the CLI, so the cached CLI profile in +//! `~/.claude.json` says nothing about who ran a desktop session. The app does +//! keep one record per session under the account that ran it: +//! `claude-code-sessions///local_.json`, +//! whose `cliSessionId` is the transcript's `sessionId`. When the app resumes a +//! conversation in a fresh CLI session, the earlier transcripts move to +//! `priorCliSessionIds`, and they belong to the same account. +//! +//! A record can also be copied into another account's directory so the chat +//! stays visible after a switch, and either copy can be continued later. Each +//! record holds the whole session up to its own `lastActivityAt`, so the +//! session splits into stretches at those instants: a stretch only one record +//! reaches belongs to that account, and the next stretch takes over once the +//! previous one's trailing usage has had time to land. A stretch several records reach is copied history; it +//! goes to the account with the latest sign-in evidence before it ended, and is +//! marked as inferred so a binding recorded before the copy existed outranks it. +//! File creation times cannot date a copy: the app rewrites records in place, +//! so a record's creation time is its last write. An account directory's +//! creation time does date that account's first sign-in. + +use super::*; +use crate::{AccountEvidenceScan, ObservedProviderAccount, CLAUDE_CODE_PROVIDER}; +use chrono::Duration; +use statsai_core::{ + conversation_account_binding_id_from, hash_text, provider_account_id_from_identity, + AccountEvidenceKind, Confidence, ConversationAccountBindingV1, SourceLocation, + CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION, CONVERSATION_ACTIVITY_GRACE_SECONDS, +}; +use std::collections::{BTreeMap, BTreeSet}; + +/// Session records are a few kilobytes; anything far larger is not one. +const MAX_DESKTOP_SESSION_RECORD_BYTES: u64 = 8 * 1024 * 1024; + +#[derive(Deserialize)] +struct DesktopSessionRecordFile { + #[serde(rename = "cliSessionId")] + cli_session_id: Option, + /// Transcripts the session ran in before the current one: the app moves a + /// conversation to a fresh CLI session when it resumes it. + #[serde(rename = "priorCliSessionIds", default)] + prior_cli_session_ids: Option>, + #[serde(rename = "createdAt")] + created_at: Option, + #[serde(rename = "lastActivityAt")] + last_activity_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct DesktopSessionRecord { + pub(crate) account_uuid: String, + /// The app's own session ID: copies of one session share it. + pub(crate) desktop_session_id: String, + /// The transcript the session runs in now. + pub(crate) cli_session_id: String, + pub(crate) prior_cli_session_ids: Vec, + pub(crate) created_at: DateTime, + pub(crate) last_activity_at: DateTime, +} + +/// The desktop sessions directory, when `root` is the configuration +/// directory the desktop app writes its transcripts into. +/// +/// Only the default `~/.claude` source can hold desktop sessions, so only it +/// reads, and watches, the desktop app's records. +pub(crate) fn claude_desktop_sessions_root_for(root: &Path) -> Option { + claude_desktop_sessions_root_in(root, &home_dir()?) +} + +pub(crate) fn claude_desktop_sessions_root_in(root: &Path, home: &Path) -> Option { + (root == home.join(".claude")) + .then(|| claude_desktop_sessions_root_from_home(home)) + .flatten() +} + +/// Where the desktop app keeps its per-account session records. +pub(crate) fn claude_desktop_sessions_root_from_home(home: &Path) -> Option { + let app_support = if cfg!(target_os = "macos") { + home.join("Library/Application Support") + } else if cfg!(windows) { + PathBuf::from(std::env::var_os("APPDATA")?) + } else { + std::env::var_os("XDG_CONFIG_HOME") + .map(PathBuf::from) + .unwrap_or_else(|| home.join(".config")) + }; + Some(app_support.join("Claude").join("claude-code-sessions")) +} + +/// Everything the desktop app's session store says about who ran what. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub(crate) struct DesktopSessions { + pub(crate) records: Vec, + /// When each account's directory appeared: its first sign-in on this + /// machine. Copying records in later cannot move it. + pub(crate) account_first_seen: BTreeMap>, + /// Sessions with a record that could not be read. Without it the copies + /// that were read would look like the only holders, so these sessions + /// yield no evidence until every copy reads cleanly. + pub(crate) incomplete_sessions: BTreeSet, + /// A directory could not be listed, so any session may be missing a copy. + pub(crate) listing_failed: bool, +} + +enum DesktopRecordRead { + Record(DesktopSessionRecord), + /// Read cleanly, but names no transcript to bind. + NoTranscript, + Unreadable, +} + +pub(crate) fn read_claude_desktop_sessions(root: &Path) -> DesktopSessions { + let mut sessions = DesktopSessions::default(); + if !root.exists() { + return sessions; + } + let Some(account_dirs) = subdirectories(root) else { + sessions.listing_failed = true; + return sessions; + }; + for account_dir in account_dirs { + let Some(account_uuid) = account_dir + .file_name() + .and_then(|name| name.to_str()) + .and_then(|name| normalized_optional_string(Some(name))) + else { + continue; + }; + if let Ok(created) = std::fs::metadata(&account_dir).and_then(|metadata| metadata.created()) + { + sessions + .account_first_seen + .insert(account_uuid.clone(), DateTime::::from(created)); + } + let Some(organization_dirs) = subdirectories(&account_dir) else { + sessions.listing_failed = true; + continue; + }; + for organization_dir in organization_dirs { + let Some(entries) = std::fs::read_dir(&organization_dir) + .ok() + .and_then(|entries| entries.collect::>>().ok()) + else { + sessions.listing_failed = true; + continue; + }; + for entry in entries { + let path = entry.path(); + let Some(desktop_session_id) = path + .file_name() + .and_then(|name| name.to_str()) + .filter(|name| name.starts_with("local_") && name.ends_with(".json")) + .and_then(|_| path.file_stem()) + .and_then(|stem| stem.to_str()) + .map(ToOwned::to_owned) + else { + continue; + }; + match read_desktop_session_record(&path, &account_uuid, &desktop_session_id) { + DesktopRecordRead::Record(record) => sessions.records.push(record), + DesktopRecordRead::NoTranscript => {} + DesktopRecordRead::Unreadable => { + sessions.incomplete_sessions.insert(desktop_session_id); + } + } + } + } + } + sessions +} + +/// The directories directly under `path`, or `None` when it cannot be listed in full. +fn subdirectories(path: &Path) -> Option> { + let mut directories = Vec::new(); + for entry in std::fs::read_dir(path).ok()? { + let entry = entry.ok()?; + if entry.file_type().ok()?.is_dir() { + directories.push(entry.path()); + } + } + directories.sort(); + Some(directories) +} + +/// One record, named by its file: copies of a session keep the file name. +fn read_desktop_session_record( + path: &Path, + account_uuid: &str, + desktop_session_id: &str, +) -> DesktopRecordRead { + let record = std::fs::metadata(path) + .ok() + .filter(|metadata| metadata.len() <= MAX_DESKTOP_SESSION_RECORD_BYTES) + .and_then(|_| File::open(path).ok()) + .and_then(|file| { + serde_json::from_reader::<_, DesktopSessionRecordFile>(BufReader::new(file)).ok() + }); + let Some(record) = record else { + return DesktopRecordRead::Unreadable; + }; + let Some(cli_session_id) = normalized_optional_string(record.cli_session_id.as_deref()) else { + return DesktopRecordRead::NoTranscript; + }; + let created_at = record.created_at.as_ref().and_then(desktop_timestamp); + let last_activity_at = record.last_activity_at.as_ref().and_then(desktop_timestamp); + // Without a time the record cannot be placed against its copies. + let Some(last_activity_at) = last_activity_at.or(created_at) else { + return DesktopRecordRead::Unreadable; + }; + let prior_cli_session_ids = record + .prior_cli_session_ids + .unwrap_or_default() + .iter() + .filter_map(|value| normalized_optional_string(value.as_str())) + .filter(|prior| *prior != cli_session_id) + .collect(); + DesktopRecordRead::Record(DesktopSessionRecord { + account_uuid: account_uuid.to_string(), + desktop_session_id: desktop_session_id.to_string(), + cli_session_id, + prior_cli_session_ids, + created_at: created_at.unwrap_or(last_activity_at), + last_activity_at, + }) +} + +fn desktop_timestamp(value: &Value) -> Option> { + let millis = value.as_i64()?; + Utc.timestamp_millis_opt(millis).single() +} + +/// One account's stretch of a session: everything up to the end of the +/// account's record that no earlier stretch already covers. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct DesktopSessionTakeover { + pub(crate) cli_session_id: String, + pub(crate) account_uuid: String, + /// `None` for the stretch the session started with. + pub(crate) active_from: Option>, + /// Where the stretch ends: the desktop app saw no activity after this, so + /// any much later usage was resumed elsewhere. + pub(crate) observed_until: Option>, + /// Where the stretch ends, when only one account's record held it. + pub(crate) certain_until: Option>, + pub(crate) observed_at: DateTime, + /// `High` when only one account's record holds the stretch; `Medium` when + /// copies held it too and the account signed in at the time was chosen. + pub(crate) confidence: Confidence, +} + +/// `(created_at, last_activity_at)` for one account's copies of one session. +type AccountSpan = (DateTime, DateTime); + +/// One session's stretches in order: where each ends, and every account whose +/// record reaches that far and so could have produced it. +fn session_stretches<'a>( + accounts: &BTreeMap<&'a str, AccountSpan>, +) -> Vec<(DateTime, Vec<&'a str>)> { + let ends = accounts + .values() + .map(|(_, last_activity_at)| *last_activity_at) + .collect::>(); + ends.into_iter() + .map(|end| { + let holders = accounts + .iter() + .filter(|(_, (_, last_activity_at))| *last_activity_at >= end) + .map(|(account_uuid, _)| *account_uuid) + .collect(); + (end, holders) + }) + .collect() +} + +/// Decide which account ran each stretch of each session. +/// +/// A record holds the whole session up to its `lastActivityAt`, so a stretch +/// held by one record is that account's for certain. A stretch several records +/// hold is history copied between accounts, and it goes to whichever holder +/// shows the latest sign-in evidence at or before the stretch ends: its +/// directory's first appearance, or a stretch only it holds. +pub(crate) fn claude_desktop_session_takeovers( + sessions: &DesktopSessions, +) -> Vec { + // Per session, per account: the earliest creation and the latest activity + // that account's records show. One account can hold a session in several + // organization directories. + if sessions.listing_failed { + return Vec::new(); + } + let mut by_session: BTreeMap<&str, BTreeMap<&str, AccountSpan>> = BTreeMap::new(); + for record in sessions.records.iter().filter(|record| { + !sessions + .incomplete_sessions + .contains(&record.desktop_session_id) + }) { + let accounts = by_session + .entry(record.desktop_session_id.as_str()) + .or_default(); + let entry = accounts + .entry(record.account_uuid.as_str()) + .or_insert((record.created_at, record.last_activity_at)); + entry.0 = entry.0.min(record.created_at); + entry.1 = entry.1.max(record.last_activity_at); + } + + let mut signed_in: BTreeMap<&str, BTreeSet>> = BTreeMap::new(); + for (account_uuid, first_seen) in &sessions.account_first_seen { + signed_in + .entry(account_uuid.as_str()) + .or_default() + .insert(*first_seen); + } + for accounts in by_session.values() { + let mut previous_end = None; + for (end, holders) in session_stretches(accounts) { + if let [account_uuid] = holders[..] { + let evidence = signed_in.entry(account_uuid).or_default(); + evidence.insert(end); + if previous_end.is_none() { + evidence.insert(accounts[account_uuid].0); + } + } + previous_end = Some(end); + } + } + + let transcripts = session_transcripts(&sessions.records); + let activity_grace = Duration::seconds(CONVERSATION_ACTIVITY_GRACE_SECONDS); + let mut takeovers = Vec::new(); + for (desktop_session_id, accounts) in &by_session { + let Some(cli_session_ids) = transcripts.get(desktop_session_id) else { + continue; + }; + let created_at = accounts + .values() + .map(|(created_at, _)| *created_at) + .min() + .expect("a session has at least one record"); + let mut previous_end: Option> = None; + for (end, holders) in session_stretches(accounts) { + // The previous account's last reply can be logged a little after + // its recorded activity, so the next stretch takes over once that + // grace has passed -- or sooner, once the next account is seen + // signed in or active, and never after the stretch itself ends. + // Nor before the next account was first seen at all: usage before + // then, such as a resume in the CLI, keeps the source's own + // attribution. + let takeover_start = |holder: Option<&str>| { + previous_end.map(|previous_end| { + let evidence = holder.and_then(|holder| signed_in.get(holder)); + let seen_after = evidence.and_then(|evidence| { + evidence + .range(( + std::ops::Bound::Excluded(previous_end), + std::ops::Bound::Unbounded, + )) + .next() + .copied() + }); + let earliest = [Some(previous_end + activity_grace), seen_after, Some(end)] + .into_iter() + .flatten() + .min() + .expect("the grace bound is always present"); + let first_seen = evidence.and_then(|evidence| evidence.first().copied()); + first_seen.map_or(earliest, |first_seen| earliest.max(first_seen).min(end)) + }) + }; + let (pieces, confidence, certain_until) = match holders[..] { + [only] => ( + vec![(takeover_start(Some(only)), only)], + Confidence::High, + Some(end), + ), + // An empty result means nothing tells the original from its + // copies: the stretch is left to the source's own attribution. + _ => ( + shared_stretch_owners( + &holders, + &signed_in, + created_at, + takeover_start(None), + end, + ), + Confidence::Medium, + None, + ), + }; + previous_end = Some(end); + // Every stretch is emitted, even when it repeats the previous + // account: a guessed stretch can be outranked by an earlier certain + // binding, and the stretch after it must still take over on its own. + for (active_from, account_uuid) in pieces { + for cli_session_id in cli_session_ids { + takeovers.push(DesktopSessionTakeover { + cli_session_id: (*cli_session_id).to_string(), + account_uuid: account_uuid.to_string(), + active_from, + observed_until: Some(end), + certain_until, + observed_at: active_from.unwrap_or(created_at), + confidence: confidence.clone(), + }); + } + } + } + } + takeovers +} + +/// Every transcript each desktop session ran in. +/// +/// A prior transcript that another session currently runs in is not this +/// session's to claim, so it is left out. +fn session_transcripts(records: &[DesktopSessionRecord]) -> BTreeMap<&str, BTreeSet<&str>> { + let mut current_owner: BTreeMap<&str, BTreeSet<&str>> = BTreeMap::new(); + for record in records { + current_owner + .entry(record.cli_session_id.as_str()) + .or_default() + .insert(record.desktop_session_id.as_str()); + } + let mut transcripts: BTreeMap<&str, BTreeSet<&str>> = BTreeMap::new(); + for record in records { + let session = transcripts + .entry(record.desktop_session_id.as_str()) + .or_default(); + session.insert(record.cli_session_id.as_str()); + for prior in &record.prior_cli_session_ids { + let claimed_elsewhere = current_owner.get(prior.as_str()).is_some_and(|owners| { + owners + .iter() + .any(|owner| *owner != record.desktop_session_id) + }); + if !claimed_elsewhere { + session.insert(prior.as_str()); + } + } + } + transcripts +} + +/// Who ran a stretch of copied history from `start` to `end`. +/// +/// The holder signed in most recently before `end` ran it, but only from when +/// that sign-in began: anything earlier belongs to whoever was signed in then. +fn shared_stretch_owners<'a>( + holders: &[&'a str], + signed_in: &BTreeMap<&str, BTreeSet>>, + session_created_at: DateTime, + start: Option>, + end: DateTime, +) -> Vec<(Option>, &'a str)> { + let Some(owner) = signed_in_at(holders, signed_in, end) else { + return Vec::new(); + }; + let others = holders + .iter() + .copied() + .filter(|holder| *holder != owner) + .collect::>(); + // The owner's current sign-in began with its first evidence after the last + // time another holder was seen signed in, not with its first sign-in ever: + // an account seen again in between ran everything up to then. + let others_last_seen = others + .iter() + .filter_map(|holder| signed_in.get(holder)?.range(..=end).next_back()) + .max() + .copied(); + let first_seen = signed_in + .get(owner) + .and_then(|evidence| match others_last_seen { + Some(after) => evidence + .range((std::ops::Bound::Excluded(after), std::ops::Bound::Unbounded)) + .next(), + None => evidence.first(), + }) + .copied() + .expect("the owner has evidence later than every other holder's"); + // Signed in by the time the stretch began: who held the account before + // then cannot matter. Before the session existed nothing could run it. + if first_seen <= start.unwrap_or(session_created_at) { + return vec![(start, owner)]; + } + // Every holder stays eligible for the earlier part: accounts can take + // turns, and the same one can own several separate stretches. + let mut owners = shared_stretch_owners( + holders, + signed_in, + session_created_at, + start, + first_seen - Duration::milliseconds(1), + ); + match owners.last() { + // Nobody is known signed in before the owner's sign-in began, so that + // earlier history stays unbound and keeps the source's own attribution. + None => vec![(Some(first_seen), owner)], + Some((_, previous)) if *previous == owner => owners, + Some(_) => { + owners.push((Some(first_seen), owner)); + owners + } + } +} + +/// The holder with the latest sign-in evidence at or before `at`, if exactly one has it. +fn signed_in_at<'a>( + holders: &[&'a str], + signed_in: &BTreeMap<&str, BTreeSet>>, + at: DateTime, +) -> Option<&'a str> { + let mut latest = holders + .iter() + .filter_map(|account_uuid| { + let evidence = signed_in.get(account_uuid)?.range(..=at).next_back()?; + Some((*evidence, *account_uuid)) + }) + .collect::>(); + latest.sort(); + match latest.as_slice() { + [] => None, + [.., (newest, account_uuid)] => { + let tied = latest.iter().filter(|(time, _)| time == newest).count(); + (tied == 1).then_some(*account_uuid) + } + } +} + +/// Bind each desktop session, or each stretch of one, to the account that ran it. +pub(crate) fn collect_claude_desktop_session_evidence( + source: &SourceLocation, + desktop_sessions_root: &Path, + scan: &mut AccountEvidenceScan, +) { + let sessions = read_claude_desktop_sessions(desktop_sessions_root); + bind_claude_desktop_sessions(source, &sessions, scan); +} + +pub(crate) fn bind_claude_desktop_sessions( + source: &SourceLocation, + sessions: &DesktopSessions, + scan: &mut AccountEvidenceScan, +) { + let mut latest_by_account: BTreeMap> = BTreeMap::new(); + for takeover in claude_desktop_session_takeovers(sessions) { + let Some(provider_account_id) = provider_account_id_from_identity( + CLAUDE_CODE_PROVIDER, + Some(&takeover.account_uuid), + None, + ) else { + continue; + }; + latest_by_account + .entry(takeover.account_uuid.clone()) + .and_modify(|latest| *latest = (*latest).max(takeover.observed_at)) + .or_insert(takeover.observed_at); + let conversation_id_hash = hash_text(&takeover.cli_session_id); + scan.conversation_bindings + .push(ConversationAccountBindingV1 { + schema_version: CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: conversation_account_binding_id_from( + &source.source_id, + &conversation_id_hash, + None, + &provider_account_id, + takeover.active_from, + ), + provider: CLAUDE_CODE_PROVIDER.to_string(), + source_id: source.source_id.clone(), + provider_account_id, + conversation_id_hash, + turn_id_hash: None, + active_from: takeover.active_from, + observed_until: takeover.observed_until, + certain_until: takeover.certain_until, + observed_at: takeover.observed_at, + // Bindings are local-only and nothing reads their kind; this is + // the app's own snapshot of which account holds the session. + evidence_kind: AccountEvidenceKind::AuthSnapshot, + confidence: takeover.confidence, + }); + } + for (account_uuid, observed_at) in latest_by_account { + scan.accounts.push(ObservedProviderAccount { + provider_user_id: Some(account_uuid), + email: None, + plan_name: None, + observed_at, + }); + } +} diff --git a/crates/statsai-adapters/src/claude/auth/mod.rs b/crates/statsai-adapters/src/claude/auth/mod.rs index 3056a9a..638db25 100644 --- a/crates/statsai-adapters/src/claude/auth/mod.rs +++ b/crates/statsai-adapters/src/claude/auth/mod.rs @@ -15,10 +15,12 @@ use std::io::BufReader; use std::path::{Path, PathBuf}; use url::Url; +mod desktop; mod overrides; mod plans; mod projects; +pub(crate) use desktop::*; pub(crate) use overrides::*; pub(crate) use plans::*; pub use projects::ClaudeProjectPathMemo; @@ -99,6 +101,12 @@ pub(crate) fn claude_auth_dependency_paths( if let Some(managed_settings_root) = claude_managed_settings_root() { paths.push(managed_settings_root); } + // Desktop session records decide which account ran each desktop session; + // the app rewrites them after the transcript, so they must trigger a rescan + // of their own. + if let Some(desktop_sessions_root) = claude_desktop_sessions_root_for(root) { + paths.push(desktop_sessions_root); + } if let Some(project_paths) = claude_project_paths_from_session_indexes(&root.join("projects")) { for project_path in project_paths { for project_settings_root in claude_project_settings_roots(&project_path) { @@ -148,7 +156,13 @@ pub(crate) fn claude_verification_dependency_topology_changed( changed: &[PathBuf], ) -> bool { let projects_root = root.join("projects"); + let desktop_sessions_root = claude_desktop_sessions_root_for(root); changed.iter().any(|path| { + // Created after the watch plan was built: it is then watched from its + // nearest existing ancestor until the plan is rebuilt recursively. + if desktop_sessions_root.as_ref() == Some(path) { + return true; + } if path == &projects_root { return true; } diff --git a/crates/statsai-adapters/src/claude/auth/plans.rs b/crates/statsai-adapters/src/claude/auth/plans.rs index 8df6a38..3936266 100644 --- a/crates/statsai-adapters/src/claude/auth/plans.rs +++ b/crates/statsai-adapters/src/claude/auth/plans.rs @@ -90,12 +90,16 @@ pub(crate) fn collect_claude_account_evidence( location_origin: &LocationOrigin, ) -> AccountEvidenceScan { let managed_settings_root = claude_managed_settings_root(); - collect_claude_account_evidence_with_probe_context( + let mut scan = collect_claude_account_evidence_with_probe_context( source, root, location_origin, managed_settings_root.as_deref(), - ) + ); + if let Some(desktop_sessions_root) = claude_desktop_sessions_root_for(root) { + collect_claude_desktop_session_evidence(source, &desktop_sessions_root, &mut scan); + } + scan } /// Collect the cached Claude subscription plan for one source. diff --git a/crates/statsai-adapters/src/claude/tests/auth/desktop.rs b/crates/statsai-adapters/src/claude/tests/auth/desktop.rs new file mode 100644 index 0000000..fd9fadb --- /dev/null +++ b/crates/statsai-adapters/src/claude/tests/auth/desktop.rs @@ -0,0 +1,734 @@ +use super::*; +use crate::{AccountEvidenceScan, CLAUDE_CODE_PROVIDER}; +use statsai_core::{ + hash_text, provider_account_id_from_identity, Confidence, CONVERSATION_ACTIVITY_GRACE_SECONDS, +}; + +const OLD_ACCOUNT: &str = "old-account-uuid"; +const NEW_ACCOUNT: &str = "new-account-uuid"; +const HOUR_MILLIS: i64 = 3_600_000; +const BASE_MILLIS: i64 = 1_786_104_000_000; + +fn at(hours: i64) -> i64 { + BASE_MILLIS + hours * HOUR_MILLIS +} + +/// When the next account takes over from a stretch that ended at `hours`. +fn after(hours: i64) -> i64 { + at(hours) + CONVERSATION_ACTIVITY_GRACE_SECONDS * 1000 +} + +fn utc(millis: i64) -> DateTime { + Utc.timestamp_millis_opt(millis) + .single() + .expect("timestamp") +} + +fn write_record( + root: &Path, + account: &str, + organization: &str, + record: &str, + cli_session_id: &str, + created_at: i64, + last_activity_at: i64, +) { + let directory = root.join(account).join(organization); + std::fs::create_dir_all(&directory).expect("organization directory"); + std::fs::write( + directory.join(format!("local_{record}.json")), + serde_json::json!({ + "sessionId": format!("local_{record}"), + "cliSessionId": cli_session_id, + "createdAt": created_at, + "lastActivityAt": last_activity_at, + "title": "ignored", + }) + .to_string(), + ) + .expect("session record"); +} + +fn record(account: &str, cli_session_id: &str, created: i64, last: i64) -> DesktopSessionRecord { + DesktopSessionRecord { + account_uuid: account.to_string(), + desktop_session_id: format!("local-{cli_session_id}"), + cli_session_id: cli_session_id.to_string(), + prior_cli_session_ids: Vec::new(), + created_at: utc(at(created)), + last_activity_at: utc(at(last)), + } +} + +/// The old account signed in long before the fixtures; the new one at hour 9. +fn sessions(records: Vec) -> DesktopSessions { + DesktopSessions { + records, + account_first_seen: [(OLD_ACCOUNT, -100), (NEW_ACCOUNT, 9)] + .into_iter() + .map(|(account, hours)| (account.to_string(), utc(at(hours)))) + .collect(), + ..DesktopSessions::default() + } +} + +type Takeover = (String, String, Option>, Confidence); + +fn takeovers(sessions: &DesktopSessions) -> Vec { + claude_desktop_session_takeovers(sessions) + .into_iter() + .map(|takeover| { + ( + takeover.cli_session_id, + takeover.account_uuid, + takeover.active_from, + takeover.confidence, + ) + }) + .collect() +} + +fn takeover( + cli_session_id: &str, + account: &str, + active_from: Option, + confidence: Confidence, +) -> Takeover { + ( + cli_session_id.to_string(), + account.to_string(), + active_from.map(utc), + confidence, + ) +} + +#[test] +fn claude_desktop_sessions_bind_to_the_account_directory_that_holds_them() { + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-old", 0, 1), + record(NEW_ACCOUNT, "cli-new", 10, 11), + ]); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-new", NEW_ACCOUNT, None, Confidence::High), + takeover("cli-old", OLD_ACCOUNT, None, Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_untouched_copies_stay_with_the_account_that_ran_them() { + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 1), + record(NEW_ACCOUNT, "cli-1", 0, 1), + record(NEW_ACCOUNT, "cli-2", 10, 11), + ]); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-1", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-2", NEW_ACCOUNT, None, Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_copy_continued_under_a_new_account_takes_over_after_the_original_ends() { + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 2), + record(NEW_ACCOUNT, "cli-1", 0, 20), + ]); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-1", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-1", NEW_ACCOUNT, Some(at(9)), Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_original_continued_after_switching_back_keeps_its_history() { + // Copied to the new account, never used there, then continued by the old + // account after switching back: the untouched copy must not claim it. + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 30), + record(NEW_ACCOUNT, "cli-1", 0, 2), + ]); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-1", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-1", OLD_ACCOUNT, Some(after(2)), Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_session_used_by_old_new_then_old_again_splits_at_first_sign_in() { + // Old ran it to hour 2, the new account (first signed in at hour 9) continued + // its copy to hour 12, and the old account continued again to hour 30. The + // old record no longer shows where its first stretch ended, so the new + // account's first sign-in bounds it. + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 30), + record(NEW_ACCOUNT, "cli-1", 0, 12), + record(NEW_ACCOUNT, "cli-2", 9, 10), + ]); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-1", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-1", NEW_ACCOUNT, Some(at(9)), Confidence::Medium), + takeover("cli-1", OLD_ACCOUNT, Some(after(12)), Confidence::High), + takeover("cli-2", NEW_ACCOUNT, None, Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_copies_without_sign_in_evidence_are_left_unbound() { + let sessions = DesktopSessions { + records: vec![ + record(OLD_ACCOUNT, "cli-1", 0, 1), + record(NEW_ACCOUNT, "cli-1", 0, 1), + ], + ..DesktopSessions::default() + }; + + assert!(takeovers(&sessions).is_empty()); +} + +#[test] +fn claude_desktop_one_account_in_several_organizations_binds_once() { + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 1), + record(OLD_ACCOUNT, "cli-1", 0, 4), + ]); + + assert_eq!( + takeovers(&sessions), + vec![takeover("cli-1", OLD_ACCOUNT, None, Confidence::High)] + ); +} + +#[test] +fn claude_desktop_skips_unreadable_and_unrelated_files() { + let dir = tempfile::tempdir().expect("tempdir"); + let organization = dir.path().join(OLD_ACCOUNT).join("org-a"); + std::fs::create_dir_all(&organization).expect("organization directory"); + std::fs::write(organization.join("local_broken.json"), "{").expect("broken"); + std::fs::write( + organization.join("local_no_cli.json"), + r#"{"createdAt": 1786104000000}"#, + ) + .expect("no cli session"); + std::fs::write(organization.join("archived-sessions.idx"), "ignored").expect("index"); + std::fs::write(dir.path().join("stray.json"), "{}").expect("stray file"); + + let sessions = read_claude_desktop_sessions(dir.path()); + assert!(sessions.records.is_empty()); + // A record that cannot be read leaves its session incomplete; one that reads + // cleanly but names no transcript does not. + assert_eq!( + sessions.incomplete_sessions, + ["local_broken".to_string()].into_iter().collect() + ); + assert!(!sessions.listing_failed); + assert_eq!( + read_claude_desktop_sessions(&dir.path().join("missing")), + DesktopSessions::default() + ); +} + +#[test] +fn claude_desktop_evidence_binds_transcript_sessions_and_reports_accounts() { + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 2), + record(NEW_ACCOUNT, "cli-1", 0, 20), + record(NEW_ACCOUNT, "cli-2", 10, 11), + ]); + let dir = tempfile::tempdir().expect("tempdir"); + let source = SourceLocation::local_adapter( + CLAUDE_CODE_PROVIDER, + "claude-code-local-jsonl", + "0", + &dir.path().join(".claude"), + LocationOrigin::Default, + ); + + let mut scan = AccountEvidenceScan::default(); + bind_claude_desktop_sessions(&source, &sessions, &mut scan); + + let account = |uuid: &str| { + provider_account_id_from_identity(CLAUDE_CODE_PROVIDER, Some(uuid), None) + .expect("account id") + }; + let mut bindings = scan + .conversation_bindings + .iter() + .map(|binding| { + assert_eq!(binding.binding_id, binding.derived_binding_id()); + assert_eq!(binding.source_id, source.source_id); + assert_eq!(binding.turn_id_hash, None); + ( + binding.conversation_id_hash.clone(), + binding.provider_account_id.clone(), + binding.active_from, + binding.confidence.clone(), + ) + }) + .collect::>(); + bindings.sort_by_key(|binding| format!("{binding:?}")); + let mut expected = vec![ + ( + hash_text("cli-1"), + account(OLD_ACCOUNT), + None, + Confidence::Medium, + ), + ( + hash_text("cli-1"), + account(NEW_ACCOUNT), + Some(utc(at(9))), + Confidence::High, + ), + ( + hash_text("cli-2"), + account(NEW_ACCOUNT), + None, + Confidence::High, + ), + ]; + expected.sort_by_key(|binding| format!("{binding:?}")); + assert_eq!(bindings, expected); + + let accounts = scan + .accounts + .iter() + .map(|observed| { + assert_eq!(observed.email, None); + observed.provider_user_id.clone().expect("account uuid") + }) + .collect::>(); + assert_eq!( + accounts, + vec![NEW_ACCOUNT.to_string(), OLD_ACCOUNT.to_string()] + ); +} + +#[test] +fn claude_desktop_reader_collects_records_from_every_account_and_organization() { + let dir = tempfile::tempdir().expect("tempdir"); + write_record(dir.path(), OLD_ACCOUNT, "org-a", "1", "cli-1", at(0), at(2)); + write_record(dir.path(), OLD_ACCOUNT, "org-b", "2", "cli-2", at(3), at(4)); + write_record( + dir.path(), + NEW_ACCOUNT, + "org-c", + "1", + "cli-1", + at(0), + at(20), + ); + + let sessions = read_claude_desktop_sessions(dir.path()); + let mut records = sessions.records.clone(); + records.sort_by(|left, right| { + (&left.account_uuid, &left.cli_session_id) + .cmp(&(&right.account_uuid, &right.cli_session_id)) + }); + + let read = |account, desktop: &str, cli, created, last| DesktopSessionRecord { + desktop_session_id: desktop.to_string(), + ..record(account, cli, created, last) + }; + assert_eq!( + records, + vec![ + read(NEW_ACCOUNT, "local_1", "cli-1", 0, 20), + read(OLD_ACCOUNT, "local_1", "cli-1", 0, 2), + read(OLD_ACCOUNT, "local_2", "cli-2", 3, 4), + ] + ); + // Directory creation times are missing on some filesystems. + if std::fs::metadata(dir.path()) + .and_then(|metadata| metadata.created()) + .is_ok() + { + assert_eq!( + sessions.account_first_seen.keys().collect::>(), + vec![NEW_ACCOUNT, OLD_ACCOUNT] + ); + } +} + +fn sessions_first_seen( + records: Vec, + first_seen: &[(&str, i64)], +) -> DesktopSessions { + DesktopSessions { + records, + account_first_seen: first_seen + .iter() + .map(|(account, hours)| (account.to_string(), utc(at(*hours)))) + .collect(), + ..DesktopSessions::default() + } +} + +#[test] +fn claude_desktop_takeover_starts_at_the_current_sign_in_not_the_first_one() { + // Both accounts predate the session. The old account was seen again at hour + // -20 (a session only it holds), after the new account's first sign-in at + // -50, so the old account started the session at hour 0 and the new account + // can only have taken over from its next evidence, the session at hour 9. + let sessions = sessions_first_seen( + vec![ + record(OLD_ACCOUNT, "cli-1", 0, 30), + record(NEW_ACCOUNT, "cli-1", 0, 12), + record(OLD_ACCOUNT, "cli-old", -20, -19), + record(NEW_ACCOUNT, "cli-new", 9, 10), + ], + &[(OLD_ACCOUNT, -100), (NEW_ACCOUNT, -50)], + ); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-1", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-1", NEW_ACCOUNT, Some(at(9)), Confidence::Medium), + takeover("cli-1", OLD_ACCOUNT, Some(after(12)), Confidence::High), + takeover("cli-new", NEW_ACCOUNT, None, Confidence::High), + takeover("cli-old", OLD_ACCOUNT, None, Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_session_started_after_a_later_first_sign_in_belongs_to_that_account() { + // Both accounts predate the session and nothing shows the old account signed + // in again after the new one's first sign-in at -50. The new account was the + // one signed in when the session began, so it ran the copied stretch; the + // old account only holds what came after its record outgrew the copy. + let sessions = sessions_first_seen( + vec![ + record(OLD_ACCOUNT, "cli-1", 0, 30), + record(NEW_ACCOUNT, "cli-1", 0, 12), + record(NEW_ACCOUNT, "cli-new", 9, 10), + ], + &[(OLD_ACCOUNT, -100), (NEW_ACCOUNT, -50)], + ); + + assert_eq!( + takeovers(&sessions), + vec![ + takeover("cli-1", NEW_ACCOUNT, None, Confidence::Medium), + takeover("cli-1", OLD_ACCOUNT, Some(after(12)), Confidence::High), + takeover("cli-new", NEW_ACCOUNT, None, Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_accounts_taking_turns_each_keep_their_stretches() { + // Sole-account sessions show B at hour 9, A at 18 and B again at 27; the + // shared session's copied history must alternate the same way. + let sessions = sessions_first_seen( + vec![ + record(OLD_ACCOUNT, "cli-1", 0, 30), + record(NEW_ACCOUNT, "cli-1", 0, 40), + record(NEW_ACCOUNT, "cli-b1", 9, 10), + record(OLD_ACCOUNT, "cli-a1", 18, 19), + record(NEW_ACCOUNT, "cli-b2", 27, 28), + ], + &[(OLD_ACCOUNT, -100), (NEW_ACCOUNT, 5)], + ); + + let shared = takeovers(&sessions) + .into_iter() + .filter(|(cli, ..)| cli == "cli-1") + .collect::>(); + assert_eq!( + shared, + vec![ + takeover("cli-1", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-1", NEW_ACCOUNT, Some(at(5)), Confidence::Medium), + takeover("cli-1", OLD_ACCOUNT, Some(at(18)), Confidence::Medium), + takeover("cli-1", NEW_ACCOUNT, Some(at(27)), Confidence::Medium), + takeover("cli-1", NEW_ACCOUNT, Some(after(30)), Confidence::High), + ] + ); +} + +#[test] +fn claude_desktop_every_stretch_records_where_its_activity_ends() { + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 2), + record(NEW_ACCOUNT, "cli-1", 0, 20), + ]); + + let ranges = claude_desktop_session_takeovers(&sessions) + .into_iter() + .map(|takeover| { + ( + takeover.confidence, + takeover.observed_until, + takeover.certain_until, + ) + }) + .collect::>(); + assert_eq!( + ranges, + vec![ + (Confidence::Medium, Some(utc(at(2))), None), + (Confidence::High, Some(utc(at(20))), Some(utc(at(20)))), + ] + ); +} + +#[test] +fn claude_desktop_reader_keeps_prior_transcripts() { + let dir = tempfile::tempdir().expect("tempdir"); + let organization = dir.path().join(NEW_ACCOUNT).join("org-a"); + std::fs::create_dir_all(&organization).expect("organization directory"); + std::fs::write( + organization.join("local_1.json"), + serde_json::json!({ + "sessionId": "local_1", + "cliSessionId": "cli-current", + "priorCliSessionIds": ["cli-first", 7, "", "cli-current", "cli-second"], + "createdAt": at(0), + "lastActivityAt": at(5), + }) + .to_string(), + ) + .expect("session record"); + + let sessions = read_claude_desktop_sessions(dir.path()); + + assert_eq!( + sessions.records, + vec![DesktopSessionRecord { + desktop_session_id: "local_1".to_string(), + prior_cli_session_ids: vec!["cli-first".to_string(), "cli-second".to_string()], + ..record(NEW_ACCOUNT, "cli-current", 0, 5) + }] + ); +} + +#[test] +fn claude_desktop_resumed_session_binds_every_transcript_it_ran_in() { + // The new account continued a copied session, and the app resumed it in a + // fresh transcript; the old account's copy still names the first one. + let mut continued = record(NEW_ACCOUNT, "cli-resumed", 0, 20); + continued.desktop_session_id = "local-shared".to_string(); + continued.prior_cli_session_ids = vec!["cli-first".to_string()]; + let mut copy = record(OLD_ACCOUNT, "cli-first", 0, 2); + copy.desktop_session_id = "local-shared".to_string(); + // A prior transcript another session currently runs in stays with it. + let mut fork = record(NEW_ACCOUNT, "cli-fork", 10, 11); + fork.prior_cli_session_ids = vec!["cli-other".to_string()]; + let other = record(OLD_ACCOUNT, "cli-other", 3, 4); + let sessions = sessions(vec![continued, copy, fork, other]); + + let mut takeovers = takeovers(&sessions); + takeovers.sort_by_key(|takeover| format!("{takeover:?}")); + let mut expected = vec![ + takeover("cli-first", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-resumed", OLD_ACCOUNT, None, Confidence::Medium), + takeover("cli-first", NEW_ACCOUNT, Some(at(9)), Confidence::High), + takeover("cli-resumed", NEW_ACCOUNT, Some(at(9)), Confidence::High), + takeover("cli-fork", NEW_ACCOUNT, None, Confidence::High), + takeover("cli-other", OLD_ACCOUNT, None, Confidence::High), + ]; + expected.sort_by_key(|takeover| format!("{takeover:?}")); + assert_eq!(takeovers, expected); +} + +#[test] +fn claude_desktop_copied_history_before_any_known_sign_in_stays_unbound() { + // Only the new account has sign-in evidence, from hour 9. The copied + // stretch from hour 0 is the new account's only from then on. + let sessions = sessions_first_seen( + vec![ + record(OLD_ACCOUNT, "cli-1", 0, 12), + record(NEW_ACCOUNT, "cli-1", 0, 12), + ], + &[(NEW_ACCOUNT, 9)], + ); + + assert_eq!( + takeovers(&sessions), + vec![takeover( + "cli-1", + NEW_ACCOUNT, + Some(at(9)), + Confidence::Medium + )] + ); +} + +#[test] +fn claude_desktop_sessions_belong_only_to_the_default_configuration_directory() { + let home = Path::new("/home/someone"); + let sessions_root = + claude_desktop_sessions_root_from_home(home).expect("desktop sessions root"); + + assert_eq!( + claude_desktop_sessions_root_in(&home.join(".claude"), home), + Some(sessions_root) + ); + assert_eq!( + claude_desktop_sessions_root_in(&home.join(".config/claude"), home), + None + ); + assert_eq!( + claude_desktop_sessions_root_in(Path::new("/elsewhere/.claude"), home), + None + ); +} + +#[test] +fn claude_auth_dependencies_omit_desktop_sessions_for_other_sources() { + let dir = tempfile::tempdir().expect("tempdir"); + let root = dir.path().join("claude-config"); + std::fs::create_dir_all(root.join("projects")).expect("projects"); + + let dependencies = claude_auth_dependency_paths(&root, &LocationOrigin::Configured); + + assert!(!dependencies + .iter() + .any(|path| path.ends_with("claude-code-sessions"))); + assert!(!claude_verification_dependency_topology_changed( + &root, + &[dir.path().join("claude-code-sessions")] + )); +} + +#[test] +fn claude_desktop_session_with_an_unreadable_copy_yields_no_evidence() { + // The old account's copy could not be read this scan: the new account's copy + // must not look like the only holder, or it would bind for certain. + let mut sessions = sessions(vec![ + record(NEW_ACCOUNT, "cli-1", 0, 20), + record(OLD_ACCOUNT, "cli-2", 3, 4), + ]); + sessions + .incomplete_sessions + .insert("local-cli-1".to_string()); + + assert_eq!( + takeovers(&sessions), + vec![takeover("cli-2", OLD_ACCOUNT, None, Confidence::High)] + ); +} + +#[test] +fn claude_desktop_failed_listing_yields_no_evidence() { + let mut sessions = sessions(vec![record(NEW_ACCOUNT, "cli-1", 0, 20)]); + sessions.listing_failed = true; + + assert!(takeovers(&sessions).is_empty()); +} + +#[cfg(unix)] +#[test] +fn claude_desktop_reader_reports_a_directory_it_cannot_list() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().expect("tempdir"); + write_record(dir.path(), OLD_ACCOUNT, "org-a", "1", "cli-1", at(0), at(2)); + let locked = dir.path().join(NEW_ACCOUNT).join("org-b"); + std::fs::create_dir_all(&locked).expect("locked organization"); + std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o000)) + .expect("lock organization"); + let unlistable = std::fs::read_dir(&locked).is_err(); + + let sessions = read_claude_desktop_sessions(dir.path()); + + std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o755)) + .expect("unlock organization"); + // Running as root can list anything; there is nothing to check then. + if unlistable { + assert!(sessions.listing_failed); + assert!(claude_desktop_session_takeovers(&sessions).is_empty()); + } +} + +#[test] +fn claude_desktop_quick_switch_takes_over_no_later_than_the_new_accounts_activity() { + let minutes = |count: i64| count * 60 * 1000; + // The new account continued the copy two minutes after the old account's + // last activity: inside the grace, so its own activity bounds the takeover. + let quick = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 2), + DesktopSessionRecord { + last_activity_at: utc(at(2) + minutes(2)), + ..record(NEW_ACCOUNT, "cli-1", 0, 2) + }, + ]); + let takeovers_with_ranges = |sessions: &DesktopSessions| { + claude_desktop_session_takeovers(sessions) + .into_iter() + .map(|takeover| { + if let (Some(from), Some(until)) = (takeover.active_from, takeover.certain_until) { + assert!(from <= until, "a range never starts after it ends"); + } + (takeover.account_uuid, takeover.active_from) + }) + .collect::>() + }; + + assert_eq!( + takeovers_with_ranges(&quick), + vec![ + (OLD_ACCOUNT.to_string(), None), + (NEW_ACCOUNT.to_string(), Some(utc(at(2) + minutes(2)))), + ] + ); + + // Seen signed in a minute after the old account stopped, through a session + // only it holds: that is when it took over. + let mut signed_in_sooner = quick.clone(); + signed_in_sooner.records.push(DesktopSessionRecord { + created_at: utc(at(2) + minutes(1)), + last_activity_at: utc(at(2) + minutes(1)), + ..record(NEW_ACCOUNT, "cli-own", 0, 0) + }); + let shared = takeovers_with_ranges(&signed_in_sooner) + .into_iter() + .take(2) + .collect::>(); + assert_eq!( + shared, + vec![ + (OLD_ACCOUNT.to_string(), None), + (NEW_ACCOUNT.to_string(), Some(utc(at(2) + minutes(1)))), + ] + ); +} + +#[test] +fn claude_desktop_takeover_never_starts_before_the_accounts_first_sign_in() { + // The old account stopped at hour 2 and the new account first signed in at + // hour 9. Usage in between, such as the session resumed in the CLI, is not + // the new account's. + let sessions = sessions(vec![ + record(OLD_ACCOUNT, "cli-1", 0, 2), + record(NEW_ACCOUNT, "cli-1", 0, 20), + ]); + + let new_account = claude_desktop_session_takeovers(&sessions) + .into_iter() + .find(|takeover| takeover.account_uuid == NEW_ACCOUNT) + .expect("new account takes over"); + assert_eq!(new_account.active_from, Some(utc(at(9)))); + assert_eq!(new_account.certain_until, Some(utc(at(20)))); +} diff --git a/crates/statsai-adapters/src/claude/tests/auth/mod.rs b/crates/statsai-adapters/src/claude/tests/auth/mod.rs index ba4ea1b..be062d4 100644 --- a/crates/statsai-adapters/src/claude/tests/auth/mod.rs +++ b/crates/statsai-adapters/src/claude/tests/auth/mod.rs @@ -1,5 +1,6 @@ pub(crate) use super::*; +mod desktop; mod overrides; mod plans; mod projects; diff --git a/crates/statsai-adapters/src/codex/telemetry.rs b/crates/statsai-adapters/src/codex/telemetry.rs index a38226b..58ed120 100644 --- a/crates/statsai-adapters/src/codex/telemetry.rs +++ b/crates/statsai-adapters/src/codex/telemetry.rs @@ -369,6 +369,9 @@ pub(crate) fn collect_codex_telemetry_evidence( provider_account_id, conversation_id_hash, turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at, evidence_kind, confidence: Confidence::High, @@ -483,6 +486,9 @@ pub(crate) fn collect_codex_reset_history_evidence( provider_account_id, conversation_id_hash, turn_id_hash, + active_from: None, + observed_until: None, + certain_until: None, observed_at, evidence_kind: AccountEvidenceKind::ResetHistory, confidence: Confidence::High, diff --git a/crates/statsai-adapters/src/lib.rs b/crates/statsai-adapters/src/lib.rs index 01d4962..ea9951a 100644 --- a/crates/statsai-adapters/src/lib.rs +++ b/crates/statsai-adapters/src/lib.rs @@ -232,12 +232,7 @@ pub fn remap_account_evidence_account_ids( for binding in &mut evidence.conversation_bindings { if let Some(canonical_id) = canonical_ids.get(&binding.provider_account_id) { binding.provider_account_id = canonical_id.clone(); - binding.binding_id = conversation_account_binding_id( - &binding.source_id, - &binding.conversation_id_hash, - binding.turn_id_hash.as_deref(), - canonical_id, - ); + binding.binding_id = binding.derived_binding_id(); } } } diff --git a/crates/statsai-core/src/account_plan.rs b/crates/statsai-core/src/account_plan.rs index 594b23a..5ee9908 100644 --- a/crates/statsai-core/src/account_plan.rs +++ b/crates/statsai-core/src/account_plan.rs @@ -103,11 +103,51 @@ pub struct ConversationAccountBindingV1 { pub provider_account_id: ProviderAccountId, pub conversation_id_hash: String, pub turn_id_hash: Option, + /// When this account took the conversation over. `None` binds the whole + /// conversation. A conversation continued under a second account carries + /// one binding per account, and each event belongs to the binding with the + /// latest `active_from` at or before it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_from: Option>, + /// How far the evidence reaches: the last activity seen under this account + /// from `active_from`. Usage more than + /// [`CONVERSATION_ACTIVITY_GRACE_SECONDS`] later is not this binding's to + /// claim, since the conversation may have been resumed elsewhere. `None` + /// sets no limit. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub observed_until: Option>, + /// How far the binding is certain: the conversation was seen held by this + /// account alone from `active_from` through this instant. A certain range + /// outranks any binding inferred later; outside it a bounded binding is + /// only an inference. Never later than `observed_until`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub certain_until: Option>, pub observed_at: DateTime, pub evidence_kind: AccountEvidenceKind, pub confidence: Confidence, } +impl ConversationAccountBindingV1 { + /// The deterministic ID for this binding's current account and locators. + #[must_use] + pub fn derived_binding_id(&self) -> String { + conversation_account_binding_id_from( + &self.source_id, + &self.conversation_id_hash, + self.turn_id_hash.as_deref(), + &self.provider_account_id, + self.active_from, + ) + } +} + +/// How long recorded conversation activity may trail the usage it covers. +/// +/// The Claude desktop app stamps a session's last activity before the reply's +/// final usage is written, by up to tens of seconds. Resuming the conversation +/// elsewhere, or under another account, takes far longer than this. +pub const CONVERSATION_ACTIVITY_GRACE_SECONDS: i64 = 5 * 60; + pub const ACCOUNT_EVIDENCE_CHECKPOINT_SCHEMA_VERSION: &str = "account_evidence_checkpoint.v1"; /// Durable high-water mark for an incrementally scanned structured evidence artifact. @@ -237,17 +277,39 @@ pub fn conversation_account_binding_id( turn_id_hash: Option<&str>, provider_account_id: &ProviderAccountId, ) -> String { - format!( - "conversation_binding_{}", - &hash_text(&format!( - "conversation_account_binding.v1:{}:{conversation_id_hash}:{}:{}", - source_id.0, - turn_id_hash.unwrap_or("none"), - provider_account_id.0 - ))[..32] + conversation_account_binding_id_from( + source_id, + conversation_id_hash, + turn_id_hash, + provider_account_id, + None, ) } +/// [`conversation_account_binding_id`] for a binding that starts at `active_from`. +/// +/// An unbounded binding keeps the ID it had before bindings could start late, +/// so existing ledger rows stay deduplicated. +#[must_use] +pub fn conversation_account_binding_id_from( + source_id: &SourceId, + conversation_id_hash: &str, + turn_id_hash: Option<&str>, + provider_account_id: &ProviderAccountId, + active_from: Option>, +) -> String { + let mut key = format!( + "conversation_account_binding.v1:{}:{conversation_id_hash}:{}:{}", + source_id.0, + turn_id_hash.unwrap_or("none"), + provider_account_id.0 + ); + if let Some(active_from) = active_from { + key.push_str(&format!(":from:{}", active_from.timestamp_millis())); + } + format!("conversation_binding_{}", &hash_text(&key)[..32]) +} + #[must_use] pub fn plan_projection_from_observation( observation: &AccountPlanObservationV1, diff --git a/crates/statsai-store/src/account_plan/evidence.rs b/crates/statsai-store/src/account_plan/evidence.rs index 38e3617..f42aa2c 100644 --- a/crates/statsai-store/src/account_plan/evidence.rs +++ b/crates/statsai-store/src/account_plan/evidence.rs @@ -159,12 +159,7 @@ impl Store { for mut binding in bindings.iter().cloned() { let previous_id = binding.binding_id.clone(); binding.provider_account_id = target_provider_account_id.clone(); - binding.binding_id = conversation_account_binding_id( - &binding.source_id, - &binding.conversation_id_hash, - binding.turn_id_hash.as_deref(), - target_provider_account_id, - ); + binding.binding_id = binding.derived_binding_id(); self.upsert_conversation_account_bindings(std::slice::from_ref(&binding))?; if previous_id != binding.binding_id { self.conn.execute( @@ -202,11 +197,32 @@ impl Store { }; let identity_ids = load_ids("account_identity_observations", "observation_id")?; let plan_ids = load_ids("account_plan_observations", "observation_id")?; - let binding_ids = load_ids("conversation_account_bindings", "binding_id")?; identity_observations .retain(|observation| !identity_ids.contains(&observation.observation_id)); plan_observations.retain(|observation| !plan_ids.contains(&observation.observation_id)); - conversation_bindings.retain(|binding| !binding_ids.contains(&binding.binding_id)); + // A known binding still counts as new when it saw more than the stored one. + let mut statement = self.conn.prepare( + "SELECT binding_id, payload FROM conversation_account_bindings WHERE source_id = ?1", + )?; + let stored_bindings = statement + .query_map([&source_id.0], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })? + .collect::>>()?; + let mut retained = Vec::with_capacity(conversation_bindings.len()); + for binding in conversation_bindings.drain(..) { + let unseen = match stored_bindings.get(&binding.binding_id) { + None => true, + Some(payload) => { + let stored: ConversationAccountBindingV1 = serde_json::from_str(payload)?; + super::observations::strengthened_binding(&stored, &binding).is_some() + } + }; + if unseen { + retained.push(binding); + } + } + *conversation_bindings = retained; Ok(()) } diff --git a/crates/statsai-store/src/account_plan/mod.rs b/crates/statsai-store/src/account_plan/mod.rs index 7be6424..44d2dee 100644 --- a/crates/statsai-store/src/account_plan/mod.rs +++ b/crates/statsai-store/src/account_plan/mod.rs @@ -5,13 +5,13 @@ use super::{ use anyhow::Result; use rusqlite::params; use statsai_core::{ - account_plan_observation_id, conversation_account_binding_id, hash_text, normalize_plan_name, - periods_overlap, plan_projection_from_observation, source_account_assignment_id, - AccountEvidenceCheckpointV1, AccountEvidenceKind, AccountEvidenceSummaryV1, - AccountIdentityObservationV1, AccountPlanObservationV1, AccountPlanProjectionV1, Confidence, - ConversationAccountBindingV1, IdentitySource, ProviderAccountId, QuotaObservationRecordV1, - SourceAccountAssignment, SourceId, UsageEvent, ACCOUNT_EVIDENCE_SUMMARY_SCHEMA_VERSION, - ACCOUNT_PLAN_OBSERVATION_SCHEMA_VERSION, SOURCE_ACCOUNT_ASSIGNMENT_SCHEMA_VERSION, + account_plan_observation_id, hash_text, normalize_plan_name, periods_overlap, + plan_projection_from_observation, source_account_assignment_id, AccountEvidenceCheckpointV1, + AccountEvidenceKind, AccountEvidenceSummaryV1, AccountIdentityObservationV1, + AccountPlanObservationV1, AccountPlanProjectionV1, Confidence, ConversationAccountBindingV1, + IdentitySource, ProviderAccountId, QuotaObservationRecordV1, SourceAccountAssignment, SourceId, + UsageEvent, ACCOUNT_EVIDENCE_SUMMARY_SCHEMA_VERSION, ACCOUNT_PLAN_OBSERVATION_SCHEMA_VERSION, + CONVERSATION_ACTIVITY_GRACE_SECONDS, SOURCE_ACCOUNT_ASSIGNMENT_SCHEMA_VERSION, }; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; @@ -101,6 +101,9 @@ mod tests { provider_account_id: account_id, conversation_id_hash: "b".repeat(64), turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at, evidence_kind: AccountEvidenceKind::ResetHistory, confidence: Confidence::High, diff --git a/crates/statsai-store/src/account_plan/observations.rs b/crates/statsai-store/src/account_plan/observations.rs index 21ff675..765dd7c 100644 --- a/crates/statsai-store/src/account_plan/observations.rs +++ b/crates/statsai-store/src/account_plan/observations.rs @@ -1,5 +1,6 @@ use super::*; use crate::{parse_rfc3339_for_row, QUOTA_PLAN_TYPE_SQL}; +use rusqlite::OptionalExtension; use std::cmp::Reverse; /// What deciding a plan run needs from a quota observation, and nothing else. @@ -428,6 +429,12 @@ impl Store { rows.map(|row| Ok(serde_json::from_str(&row?)?)).collect() } + /// Records new bindings, and strengthens known ones in place. + /// + /// A binding's ID names its account and takeover instant, so a later scan + /// can rediscover it having seen more: a longer certain range, or the + /// account holding the conversation alone where it was once only inferred. + /// That is kept; nothing a later scan sees can shrink or weaken it. pub fn upsert_conversation_account_bindings( &self, bindings: &[ConversationAccountBindingV1], @@ -437,7 +444,7 @@ impl Store { } self.with_immediate_transaction(|| { let mut written = 0u64; - let mut statement = self.conn.prepare( + let mut insert = self.conn.prepare( r#" INSERT INTO conversation_account_bindings ( binding_id, provider, source_id, provider_account_id, @@ -447,8 +454,14 @@ impl Store { ON CONFLICT(binding_id) DO NOTHING "#, )?; + let mut stored = self.conn.prepare( + "SELECT payload FROM conversation_account_bindings WHERE binding_id = ?1", + )?; + let mut update = self.conn.prepare( + "UPDATE conversation_account_bindings SET payload = ?2 WHERE binding_id = ?1", + )?; for binding in bindings { - written += statement.execute(params![ + let inserted = insert.execute(params![ &binding.binding_id, &binding.provider, &binding.source_id.0, @@ -457,7 +470,25 @@ impl Store { binding.turn_id_hash.as_deref(), binding.observed_at.to_rfc3339(), serde_json::to_string(binding)?, - ])? as u64; + ])?; + if inserted > 0 { + written += 1; + continue; + } + let payload = stored + .query_row([&binding.binding_id], |row| row.get::<_, String>(0)) + .optional()?; + let Some(payload) = payload else { + continue; + }; + let existing: ConversationAccountBindingV1 = serde_json::from_str(&payload)?; + if let Some(strengthened) = strengthened_binding(&existing, binding) { + update.execute(params![ + &binding.binding_id, + serde_json::to_string(&strengthened)?, + ])?; + written += 1; + } } Ok(written) }) @@ -488,26 +519,51 @@ impl Store { events: &mut [UsageEvent], ) -> Result<()> { let bindings = self.conversation_account_bindings(Some(source_id))?; - let mut accounts_by_conversation: HashMap<&str, HashSet<&ProviderAccountId>> = + // A turn-scoped reset-history record cannot safely relabel every event in its + // conversation. Usage events currently expose only conversation identity, so + // retain that binding as evidence until an exact turn locator is available. + let conversation_bindings = || { + bindings + .iter() + .filter(|binding| binding.turn_id_hash.is_none()) + }; + let accounts_by_conversation = effective_binding_accounts(conversation_bindings()); + let mut certain_by_conversation: HashMap<&str, Vec<&ConversationAccountBindingV1>> = HashMap::new(); - for binding in bindings - .iter() - // A turn-scoped reset-history record cannot safely relabel every event in its - // conversation. Usage events currently expose only conversation identity, so retain - // that binding as evidence until an exact turn locator is available. - .filter(|binding| binding.turn_id_hash.is_none()) - { - accounts_by_conversation + for binding in conversation_bindings().filter(|binding| binding.certain_until.is_some()) { + certain_by_conversation .entry(binding.conversation_id_hash.as_str()) .or_default() - .insert(&binding.provider_account_id); + .push(binding); } for event in events { let Some(conversation_id_hash) = event.session.local_session_id_hash.as_deref() else { continue; }; - let Some(accounts) = accounts_by_conversation.get(conversation_id_hash) else { - continue; + let at = event.session.started_at; + // Inside a range a binding saw for certain, nothing inferred later + // can relabel the event; elsewhere the latest takeover decides. + let certain = certain_by_conversation + .get(conversation_id_hash) + .and_then(|bindings| certain_accounts_at(bindings, at)); + let accounts = match certain { + Some(accounts) => accounts, + None => { + let Some((_, takeover)) = accounts_by_conversation + .get(conversation_id_hash) + .and_then(|takeovers| takeovers.range(..=Some(at)).next_back()) + else { + continue; + }; + // With every binding past its last activity, the + // conversation may have been resumed elsewhere: leave it to + // the source. + let accounts = takeover.accounts_at(at); + if accounts.is_empty() { + continue; + } + accounts + } }; if accounts.len() != 1 { // Conflicting bindings are weaker evidence than a person telling @@ -635,25 +691,25 @@ impl Store { let observations = self.account_identity_observations(None)?; let bindings = self.conversation_account_bindings(None)?; let mut conversations_by_account = HashMap::>::new(); - let mut accounts_by_conversation = HashMap::>::new(); for binding in &bindings { conversations_by_account .entry(binding.provider_account_id.clone()) .or_default() .insert(binding.conversation_id_hash.clone()); - accounts_by_conversation - .entry(binding.conversation_id_hash.clone()) - .or_default() - .insert(binding.provider_account_id.clone()); } + // A conversation another account took over part-way is not a conflict; + // two accounts claiming the same stretch with equal confidence is. let mut conflicting_conversations_by_account = HashMap::::new(); - for accounts in accounts_by_conversation - .values() - .filter(|accounts| accounts.len() > 1) - { - for account in accounts { + for takeovers in effective_binding_accounts(bindings.iter()).values() { + let conflicting = takeovers + .values() + .map(TakeoverAccounts::accounts) + .filter(|accounts| accounts.len() > 1) + .flatten() + .collect::>(); + for account in conflicting { *conflicting_conversations_by_account - .entry(account.clone()) + .entry((*account).clone()) .or_default() += 1; } } @@ -677,8 +733,27 @@ impl Store { .or_default() .push(observation); } + // An account known only from conversation bindings, such as one the + // Claude desktop app ran sessions under, still has evidence to report. + let mut binding_kinds = + BTreeMap::<(String, ProviderAccountId), BTreeSet>::new(); + for binding in &bindings { + let key = ( + binding.provider.clone(), + binding.provider_account_id.clone(), + ); + binding_kinds + .entry(key.clone()) + .or_default() + .insert(binding.evidence_kind); + grouped.entry(key).or_default(); + } let mut summaries = Vec::with_capacity(grouped.len()); for ((provider, provider_account_id), observations) in grouped { + let bound_kinds = binding_kinds + .get(&(provider.clone(), provider_account_id.clone())) + .cloned() + .unwrap_or_default(); let strong = observations .iter() .filter(|observation| observation.evidence_kind.is_strong_identity()) @@ -687,6 +762,7 @@ impl Store { let evidence_kinds = observations .iter() .map(|observation| observation.evidence_kind) + .chain(bound_kinds) .collect::>() .into_iter() .collect(); @@ -838,3 +914,126 @@ fn quota_plan_observations_from_runs( }) .collect() } + +/// Per conversation, the accounts bound from each takeover instant. +type BindingAccounts<'a> = + HashMap<&'a str, BTreeMap>, TakeoverAccounts<'a>>>; + +/// The accounts bound from one takeover instant, and how far their evidence reaches. +#[derive(Default)] +struct TakeoverAccounts<'a> { + /// Each binding's account and the last activity it saw; `None` sets no limit. + members: Vec<(&'a ProviderAccountId, Option>)>, +} + +impl<'a> TakeoverAccounts<'a> { + fn accounts(&self) -> HashSet<&'a ProviderAccountId> { + self.members.iter().map(|(account, _)| *account).collect() + } + + /// The accounts still claiming `at`. Past the last activity a binding saw, + /// the conversation may have been resumed elsewhere, so that binding drops + /// out rather than contest the others. + fn accounts_at(&self, at: chrono::DateTime) -> HashSet<&'a ProviderAccountId> { + let grace = chrono::Duration::seconds(CONVERSATION_ACTIVITY_GRACE_SECONDS); + self.members + .iter() + .filter(|(_, reach)| reach.is_none_or(|reach| at <= reach + grace)) + .map(|(account, _)| *account) + .collect() + } +} + +/// Per conversation, the accounts bound from each takeover instant. +/// +/// `None` sorts first, so an unbounded binding covers everything before the +/// first takeover. Where bindings for one instant disagree, only the most +/// confident ones count: a binding inferred from copied history must not +/// unseat one recorded while only a single account held the conversation. +/// Disagreement at the same confidence is left for the caller to treat as a +/// conflict. +fn effective_binding_accounts<'a>( + bindings: impl Iterator, +) -> BindingAccounts<'a> { + // Per (conversation, takeover instant): the best confidence seen and the + // bindings at it. + let mut strongest: HashMap<_, (u8, TakeoverAccounts<'a>)> = HashMap::new(); + for binding in bindings { + // Outside its certain range a bounded binding is only an inference, + // however confident the range itself was; callers consult certain + // ranges first. + let rank = if binding.observed_until.is_some() { + confidence_rank(&Confidence::Medium) + } else { + confidence_rank(&binding.confidence) + }; + let entry = strongest + .entry((binding.conversation_id_hash.as_str(), binding.active_from)) + .or_insert_with(|| (rank, TakeoverAccounts::default())); + if rank > entry.0 { + *entry = (rank, TakeoverAccounts::default()); + } + if rank == entry.0 { + entry + .1 + .members + .push((&binding.provider_account_id, binding.observed_until)); + } + } + let mut accounts_by_conversation = BindingAccounts::new(); + for ((conversation_id_hash, active_from), (_, takeover)) in strongest { + accounts_by_conversation + .entry(conversation_id_hash) + .or_default() + .insert(active_from, takeover); + } + accounts_by_conversation +} + +/// The accounts whose certain range covers `at`, taking the latest-starting +/// range when several do, or `None` when no certain range covers it. +fn certain_accounts_at<'a>( + bindings: &[&'a ConversationAccountBindingV1], + at: chrono::DateTime, +) -> Option> { + let covering = bindings + .iter() + .filter(|binding| binding.active_from.is_none_or(|from| from <= at)) + .filter(|binding| binding.certain_until.is_some_and(|until| at <= until)) + .collect::>(); + let latest = covering.iter().map(|binding| binding.active_from).max()?; + Some( + covering + .into_iter() + .filter(|binding| binding.active_from == latest) + .map(|binding| &binding.provider_account_id) + .collect(), + ) +} + +/// `stored` with whatever more `incoming` saw, or `None` when it saw nothing more. +/// +/// Only the certain range and the confidence can grow; everything else is the +/// ledger's first record. +pub(super) fn strengthened_binding( + stored: &ConversationAccountBindingV1, + incoming: &ConversationAccountBindingV1, +) -> Option { + let mut strengthened = stored.clone(); + if confidence_rank(&incoming.confidence) > confidence_rank(&stored.confidence) { + strengthened.confidence = incoming.confidence.clone(); + } + // `None` sorts first, so a range only ever extends. The two ranges merge + // separately: an inference reaching further never makes it certain. + strengthened.observed_until = stored.observed_until.max(incoming.observed_until); + strengthened.certain_until = stored.certain_until.max(incoming.certain_until); + (strengthened != *stored).then_some(strengthened) +} + +fn confidence_rank(confidence: &Confidence) -> u8 { + match confidence { + Confidence::Low => 0, + Confidence::Medium => 1, + Confidence::High => 2, + } +} diff --git a/crates/statsai-store/src/tests/verified.rs b/crates/statsai-store/src/tests/verified.rs index 84c2c3a..49ffd9c 100644 --- a/crates/statsai-store/src/tests/verified.rs +++ b/crates/statsai-store/src/tests/verified.rs @@ -702,6 +702,9 @@ fn direct_conversation_evidence_overrides_only_that_event_and_preserves_manual_i provider_account_id: directly_bound_account.clone(), conversation_id_hash: "same-session".to_string(), turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at, evidence_kind: statsai_core::AccountEvidenceKind::ResetHistory, confidence: Confidence::High, @@ -914,6 +917,9 @@ fn conflicting_bindings_keep_a_manual_account_assignment() { provider_account_id: ProviderAccountId((*account).to_string()), conversation_id_hash: "same-session".to_string(), turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at: now, evidence_kind: statsai_core::AccountEvidenceKind::TelemetryIdentity, confidence: Confidence::High, @@ -1200,3 +1206,714 @@ fn refreshing_verified_at_does_not_rewrite_unchanged_source_records() { .expect("verified assignment"); assert_eq!(assignment.verified_at, Some(refreshed_at)); } + +#[test] +fn conversation_takeover_bindings_split_one_conversation_by_event_time() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-takeover"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let takeover = start + chrono::Duration::hours(2); + let original_account = ProviderAccountId("original-account".to_string()); + let new_account = ProviderAccountId("new-account".to_string()); + let binding = |account: &ProviderAccountId, active_from| { + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: account.clone(), + conversation_id_hash: "same-session".to_string(), + turn_id_hash: None, + active_from, + observed_until: None, + certain_until: None, + observed_at: active_from.unwrap_or(start), + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence: Confidence::High, + }; + binding.binding_id = binding.derived_binding_id(); + binding + }; + let bindings = [ + binding(&original_account, None), + binding(&new_account, Some(takeover)), + ]; + assert_ne!(bindings[0].binding_id, bindings[1].binding_id); + assert_eq!( + store + .upsert_conversation_account_bindings(&bindings) + .expect("bindings"), + 2 + ); + let mut events = vec![ + test_store_event(&source, start, "before"), + test_store_event( + &source, + takeover - chrono::Duration::milliseconds(1), + "edge", + ), + test_store_event(&source, takeover, "at"), + test_store_event(&source, takeover + chrono::Duration::hours(1), "after"), + ]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + let accounts = events + .iter() + .map(|event| event.provider_account_id.clone()) + .collect::>(); + assert_eq!( + accounts, + vec![ + Some(original_account.clone()), + Some(original_account), + Some(new_account.clone()), + Some(new_account), + ] + ); +} + +#[test] +fn conversation_takeover_binding_leaves_earlier_events_to_the_source() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-late-takeover"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let takeover = start + chrono::Duration::hours(2); + let source_account = ProviderAccountId("source-account".to_string()); + let new_account = ProviderAccountId("new-account".to_string()); + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: new_account.clone(), + conversation_id_hash: "same-session".to_string(), + turn_id_hash: None, + active_from: Some(takeover), + observed_until: None, + certain_until: None, + observed_at: takeover, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence: Confidence::High, + }; + binding.binding_id = binding.derived_binding_id(); + store + .upsert_conversation_account_bindings(&[binding]) + .expect("binding"); + let mut before = test_store_event(&source, start, "before"); + before.provider_account_id = Some(source_account.clone()); + let mut after = test_store_event(&source, takeover, "after"); + after.provider_account_id = Some(source_account.clone()); + let mut events = vec![before, after]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply binding"); + + assert_eq!(events[0].provider_account_id, Some(source_account)); + assert_eq!(events[1].provider_account_id, Some(new_account)); +} + +#[test] +fn certain_conversation_binding_outranks_one_inferred_from_copied_history() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-confidence"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let original_account = ProviderAccountId("original-account".to_string()); + let copy_account = ProviderAccountId("copy-account".to_string()); + let binding = |conversation: &str, account: &ProviderAccountId, confidence| { + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: account.clone(), + conversation_id_hash: conversation.to_string(), + turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, + observed_at: start, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence, + }; + binding.binding_id = binding.derived_binding_id(); + binding + }; + store + .upsert_conversation_account_bindings(&[ + binding("ranked", &original_account, Confidence::High), + binding("ranked", ©_account, Confidence::Medium), + binding("tied", &original_account, Confidence::Medium), + binding("tied", ©_account, Confidence::Medium), + ]) + .expect("bindings"); + let mut ranked = test_store_event(&source, start, "ranked"); + ranked.session.local_session_id_hash = Some("ranked".to_string()); + let mut tied = test_store_event(&source, start, "tied"); + tied.session.local_session_id_hash = Some("tied".to_string()); + tied.provider_account_id = Some(original_account.clone()); + let mut events = vec![ranked, tied]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + assert_eq!(events[0].provider_account_id, Some(original_account)); + assert_eq!(events[1].provider_account_id, None); +} + +#[test] +fn certain_range_outranks_a_later_inferred_takeover() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-certain-range"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let hours = chrono::Duration::hours; + let original_account = ProviderAccountId("original-account".to_string()); + let inferred_account = ProviderAccountId("inferred-account".to_string()); + let binding = |account: &ProviderAccountId, active_from, observed_until, confidence| { + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: account.clone(), + conversation_id_hash: "same-session".to_string(), + turn_id_hash: None, + active_from, + observed_until, + // A direct observation is certain through its whole range. + certain_until: if confidence == Confidence::High { + observed_until + } else { + None + }, + observed_at: start, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence, + }; + binding.binding_id = binding.derived_binding_id(); + binding + }; + // Seen held by the original account alone through hour 2; a later scan of + // copied history inferred the other account from a sign-in before the + // session began. + store + .upsert_conversation_account_bindings(&[ + binding( + &original_account, + None, + Some(start + hours(2)), + Confidence::High, + ), + binding( + &inferred_account, + Some(start - hours(50)), + None, + Confidence::Medium, + ), + ]) + .expect("bindings"); + let mut events = vec![ + test_store_event(&source, start + hours(1), "certain"), + test_store_event(&source, start + hours(2), "certain-edge"), + test_store_event(&source, start + hours(5), "inferred"), + ]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + let accounts = events + .iter() + .map(|event| event.provider_account_id.clone()) + .collect::>(); + assert_eq!( + accounts, + vec![ + Some(original_account.clone()), + Some(original_account), + Some(inferred_account), + ] + ); +} + +#[test] +fn rediscovered_binding_extends_its_certain_range_and_never_weakens() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-range-extension"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let hours = chrono::Duration::hours; + let original_account = ProviderAccountId("original-account".to_string()); + let inferred_account = ProviderAccountId("inferred-account".to_string()); + let binding = |account: &ProviderAccountId, active_from, observed_until, confidence| { + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: account.clone(), + conversation_id_hash: "same-session".to_string(), + turn_id_hash: None, + active_from, + observed_until, + // A direct observation is certain through its whole range. + certain_until: if confidence == Confidence::High { + observed_until + } else { + None + }, + observed_at: start, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence, + }; + binding.binding_id = binding.derived_binding_id(); + binding + }; + let seen_through = |hour| { + binding( + &original_account, + None, + Some(start + hours(hour)), + Confidence::High, + ) + }; + assert_eq!( + store + .upsert_conversation_account_bindings(&[seen_through(2)]) + .expect("first scan"), + 1 + ); + + // A later scan saw the original account alone through hour 5. + let mut extended = vec![seen_through(5)]; + store + .retain_unseen_account_evidence( + &source.source_id, + &mut Vec::new(), + &mut Vec::new(), + &mut extended, + ) + .expect("retain"); + assert_eq!(extended.len(), 1, "a longer certain range is new evidence"); + assert_eq!( + store + .upsert_conversation_account_bindings(&extended) + .expect("extend"), + 1 + ); + + // Neither a repeat nor a weaker rediscovery changes it. + let weaker = binding(&original_account, None, None, Confidence::Medium); + let mut rediscovered = vec![seen_through(5), seen_through(3), weaker.clone()]; + store + .retain_unseen_account_evidence( + &source.source_id, + &mut Vec::new(), + &mut Vec::new(), + &mut rediscovered, + ) + .expect("retain"); + assert!(rediscovered.is_empty()); + assert_eq!( + store + .upsert_conversation_account_bindings(&[weaker]) + .expect("weaker"), + 0 + ); + + // Copied history later inferred the other account from hour 3. + store + .upsert_conversation_account_bindings(&[binding( + &inferred_account, + Some(start + hours(3)), + None, + Confidence::Medium, + )]) + .expect("inferred"); + let mut events = vec![ + test_store_event(&source, start + hours(4), "certain"), + test_store_event(&source, start + hours(6), "inferred"), + ]; + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + assert_eq!(events[0].provider_account_id, Some(original_account)); + assert_eq!(events[1].provider_account_id, Some(inferred_account)); +} + +#[test] +fn evidence_summaries_include_accounts_known_only_from_bindings() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-summary"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let original_account = ProviderAccountId("original-account".to_string()); + let desktop_account = ProviderAccountId("desktop-account".to_string()); + let binding = |conversation: &str, account: &ProviderAccountId, active_from| { + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: account.clone(), + conversation_id_hash: conversation.to_string(), + turn_id_hash: None, + active_from, + observed_until: None, + certain_until: None, + observed_at: start, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence: Confidence::High, + }; + binding.binding_id = binding.derived_binding_id(); + binding + }; + // The desktop account took one session over part-way and ran another alone. + store + .upsert_conversation_account_bindings(&[ + binding("taken-over", &original_account, None), + binding( + "taken-over", + &desktop_account, + Some(start + chrono::Duration::hours(2)), + ), + binding("own", &desktop_account, None), + ]) + .expect("bindings"); + + let summaries = store + .account_evidence_summaries("device") + .expect("summaries"); + + let desktop = summaries + .iter() + .find(|summary| summary.provider_account_id == desktop_account) + .expect("binding-only account is summarized"); + assert_eq!(desktop.provider, "claude_code"); + assert_eq!(desktop.strong_observation_count, 0); + assert_eq!(desktop.directly_bound_conversations, 2); + assert_eq!(desktop.conflict_count, 0); + assert_eq!( + desktop.evidence_kinds, + vec![statsai_core::AccountEvidenceKind::AuthSnapshot] + ); + assert!(summaries + .iter() + .any(|summary| summary.provider_account_id == original_account)); +} + +#[test] +fn binding_stops_after_its_recorded_activity_so_a_resume_elsewhere_keeps_the_source() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-resume-elsewhere"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let desktop_account = ProviderAccountId("desktop-account".to_string()); + let cli_account = ProviderAccountId("cli-account".to_string()); + let last_desktop_activity = start + chrono::Duration::hours(4); + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: desktop_account.clone(), + conversation_id_hash: "same-session".to_string(), + turn_id_hash: None, + active_from: None, + observed_until: Some(last_desktop_activity), + certain_until: None, + observed_at: start, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence: Confidence::High, + }; + binding.binding_id = binding.derived_binding_id(); + store + .upsert_conversation_account_bindings(&[binding]) + .expect("binding"); + let event = |at, record: &str| { + let mut event = test_store_event(&source, at, record); + // What the source's own attribution gave it. + event.provider_account_id = Some(cli_account.clone()); + event + }; + let mut events = vec![ + event(last_desktop_activity, "desktop"), + // The app records activity a little before the reply's final usage. + event( + last_desktop_activity + chrono::Duration::seconds(40), + "desktop-lag", + ), + // Resumed in the CLI an hour later. + event(last_desktop_activity + chrono::Duration::hours(1), "cli"), + ]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply binding"); + + let accounts = events + .iter() + .map(|event| event.provider_account_id.clone()) + .collect::>(); + assert_eq!( + accounts, + vec![ + Some(desktop_account.clone()), + Some(desktop_account), + Some(cli_account), + ] + ); +} + +fn desktop_binding( + source: &SourceLocation, + account: &ProviderAccountId, + active_from: Option>, + observed_until: chrono::DateTime, + certain: bool, +) -> statsai_core::ConversationAccountBindingV1 { + let mut binding = statsai_core::ConversationAccountBindingV1 { + schema_version: statsai_core::CONVERSATION_ACCOUNT_BINDING_SCHEMA_VERSION.to_string(), + binding_id: String::new(), + provider: "claude_code".to_string(), + source_id: source.source_id.clone(), + provider_account_id: account.clone(), + conversation_id_hash: "same-session".to_string(), + turn_id_hash: None, + active_from, + observed_until: Some(observed_until), + certain_until: certain.then_some(observed_until), + observed_at: observed_until, + evidence_kind: statsai_core::AccountEvidenceKind::AuthSnapshot, + confidence: if certain { + Confidence::High + } else { + Confidence::Medium + }, + }; + binding.binding_id = binding.derived_binding_id(); + binding +} + +#[test] +fn an_inferred_reach_never_extends_a_certain_range() { + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let hours = chrono::Duration::hours; + let original_account = ProviderAccountId("original-account".to_string()); + let other_account = ProviderAccountId("other-account".to_string()); + for certain_first in [true, false] { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-merged-ranges"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + // Seen alone through hour 2, and inferred to hour 12 from copied history + // that the other account actually ran from hour 9. + let certain = desktop_binding(&source, &original_account, None, start + hours(2), true); + let inferred = desktop_binding(&source, &original_account, None, start + hours(12), false); + let ordered = if certain_first { + [certain, inferred] + } else { + [inferred, certain] + }; + for binding in ordered { + store + .upsert_conversation_account_bindings(&[binding]) + .expect("binding"); + } + store + .upsert_conversation_account_bindings(&[desktop_binding( + &source, + &other_account, + Some(start + hours(9)), + start + hours(12), + false, + )]) + .expect("other binding"); + let mut events = vec![ + test_store_event(&source, start + hours(1), "certain"), + test_store_event(&source, start + hours(10), "inferred"), + ]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + assert_eq!( + events[0].provider_account_id, + Some(original_account.clone()), + "certain first: {certain_first}" + ); + assert_eq!( + events[1].provider_account_id, + Some(other_account.clone()), + "certain first: {certain_first}" + ); + } +} + +#[test] +fn the_previous_accounts_trailing_reply_stays_with_it_across_a_takeover() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-trailing-reply"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let hours = chrono::Duration::hours; + let grace = chrono::Duration::seconds(statsai_core::CONVERSATION_ACTIVITY_GRACE_SECONDS); + let previous_account = ProviderAccountId("previous-account".to_string()); + let next_account = ProviderAccountId("next-account".to_string()); + let previous_end = start + hours(2); + store + .upsert_conversation_account_bindings(&[ + desktop_binding(&source, &previous_account, None, previous_end, true), + desktop_binding( + &source, + &next_account, + Some(previous_end + grace), + start + hours(20), + true, + ), + ]) + .expect("bindings"); + let mut events = vec![ + test_store_event( + &source, + previous_end + chrono::Duration::seconds(40), + "late", + ), + test_store_event(&source, start + hours(3), "next"), + ]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + assert_eq!(events[0].provider_account_id, Some(previous_account)); + assert_eq!(events[1].provider_account_id, Some(next_account)); +} + +#[test] +fn an_expired_binding_does_not_contest_one_still_in_reach() { + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "claude_code", + "test", + "0", + Path::new("/tmp/claude-desktop-expired-peer"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let start = Utc + .with_ymd_and_hms(2026, 8, 1, 0, 0, 0) + .single() + .expect("start"); + let hours = chrono::Duration::hours; + let expired_account = ProviderAccountId("expired-account".to_string()); + let current_account = ProviderAccountId("current-account".to_string()); + // Both inferred from the same instant; one saw activity only to hour 1. + store + .upsert_conversation_account_bindings(&[ + desktop_binding(&source, &expired_account, None, start + hours(1), false), + desktop_binding(&source, ¤t_account, None, start + hours(10), false), + ]) + .expect("bindings"); + let mut events = vec![ + test_store_event(&source, start, "contested"), + test_store_event(&source, start + hours(5), "current"), + ]; + + store + .apply_conversation_account_bindings(&source.source_id, &mut events) + .expect("apply bindings"); + + assert_eq!(events[0].provider_account_id, None); + assert_eq!(events[1].provider_account_id, Some(current_account)); +} diff --git a/crates/statsai/src/cli/tests/account/merge.rs b/crates/statsai/src/cli/tests/account/merge.rs index 53784ec..e45fa6c 100644 --- a/crates/statsai/src/cli/tests/account/merge.rs +++ b/crates/statsai/src/cli/tests/account/merge.rs @@ -129,6 +129,9 @@ fn merge_provider_accounts_moves_source_records_and_prunes_alias() { provider_account_id: alias.provider_account_id.clone(), conversation_id_hash: "b".repeat(64), turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at: now, evidence_kind: statsai_core::AccountEvidenceKind::ResetHistory, confidence: Confidence::High, diff --git a/crates/statsai/src/cli/tests/account/mod.rs b/crates/statsai/src/cli/tests/account/mod.rs index 432135f..6ab0231 100644 --- a/crates/statsai/src/cli/tests/account/mod.rs +++ b/crates/statsai/src/cli/tests/account/mod.rs @@ -237,6 +237,9 @@ fn known_account_aliases_are_applied_before_evidence_deduplication() { provider_account_id: raw_account_id, conversation_id_hash: "c".repeat(64), turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at, evidence_kind: statsai_core::AccountEvidenceKind::ResetHistory, confidence: Confidence::High, diff --git a/crates/statsai/src/cli/tests/source/mod.rs b/crates/statsai/src/cli/tests/source/mod.rs index 7c40f64..5014a60 100644 --- a/crates/statsai/src/cli/tests/source/mod.rs +++ b/crates/statsai/src/cli/tests/source/mod.rs @@ -106,6 +106,9 @@ fn seed_test_account_evidence(store: &Store, source: &SourceLocation, observed_a provider_account_id: account_id, conversation_id_hash: "b".repeat(64), turn_id_hash: None, + active_from: None, + observed_until: None, + certain_until: None, observed_at, evidence_kind: statsai_core::AccountEvidenceKind::ResetHistory, confidence: Confidence::High,