From 6bc2ea4eb893ebd11316743292abac46ffc319ff Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 18:13:30 +0300 Subject: [PATCH 1/7] Revoke the device session on the server at logout statsai auth logout now posts the stored refresh token to /api/devices/logout before clearing local credentials. The call is best effort: a network error or non-2xx response prints a warning and local credentials are still cleared. --local-only skips the server call. --- README.md | 6 + crates/statsai/src/auth/logout.rs | 66 +++++++++- crates/statsai/src/auth/tests.rs | 151 +++++++++++++++++++++- crates/statsai/src/cli/args/mod.rs | 12 +- crates/statsai/src/cli/auth.rs | 2 +- crates/statsai/tests/cli_surface/help.txt | 5 +- 6 files changed, 234 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index af87a47..36b8d1f 100644 --- a/README.md +++ b/README.md @@ -457,6 +457,8 @@ statsai auth login statsai auth login --no-open statsai auth login --headless --device-name "Mini server" statsai auth status +statsai auth logout +statsai auth logout --local-only statsai sync --sink file --output ./statsai-sync-batch.json statsai sync --sink http --since-last statsai sync --sink http --verify @@ -467,6 +469,10 @@ statsai schema cache-report HTTP sync uses the stored device session unless `--auth-token` or `STATSAI_SYNC_TOKEN` is provided. Access tokens are refreshed automatically. +`statsai auth logout` revokes this device's session on the server before +clearing local credentials; if the server cannot be reached it still logs out +locally, and the session expires within 30 days or can be revoked from the +dashboard's Devices page. `--local-only` skips the server call. diff --git a/crates/statsai/src/auth/logout.rs b/crates/statsai/src/auth/logout.rs index 7d0040d..a9452f2 100644 --- a/crates/statsai/src/auth/logout.rs +++ b/crates/statsai/src/auth/logout.rs @@ -1,8 +1,19 @@ use super::*; -pub fn logout() -> Result<()> { +pub fn logout(local_only: bool) -> Result<()> { let api_base_url = cloudflare_api_url(); - if logout_backend(&auth_base_dir(), &api_base_url, |api_base_url| { + let base = auth_base_dir(); + if !local_only { + if let ServerRevocation::Failed(reason) = + revoke_server_session(&base, &api_base_url, post_device_logout) + { + eprintln!("Warning: could not revoke this device's session on the server: {reason}"); + eprintln!( + "It expires within 30 days, or you can revoke it now from the Devices page of the dashboard." + ); + } + } + if logout_backend(&base, &api_base_url, |api_base_url| { delete_tokens_from_keyring_for_api_base_url(api_base_url) })? { println!("Successfully logged out."); @@ -12,6 +23,57 @@ pub fn logout() -> Result<()> { Ok(()) } +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum ServerRevocation { + /// No refresh token is stored, so there is no server session to revoke. + NotAttempted, + Revoked, + Failed(String), +} + +/// Best-effort server-side revocation of the stored device session. Never +/// fails: the caller clears local credentials whatever the outcome. +pub(crate) fn revoke_server_session( + base: &Path, + api_base_url: &str, + send: impl FnOnce(&str, &str) -> Result<()>, +) -> ServerRevocation { + let refresh_token = match stored_refresh_token(base, api_base_url) { + Ok(Some(token)) => token, + Ok(None) => return ServerRevocation::NotAttempted, + Err(error) => return ServerRevocation::Failed(format!("{error:#}")), + }; + match send(api_base_url, &refresh_token) { + Ok(()) => ServerRevocation::Revoked, + Err(error) => ServerRevocation::Failed(format!("{error:#}")), + } +} + +fn stored_refresh_token(base: &Path, api_base_url: &str) -> Result> { + if !auth_record_candidate_exists(base, api_base_url)? { + return Ok(None); + } + let Some((_path, credentials)) = auth_record_for_backend(base, api_base_url)? else { + return Ok(None); + }; + Ok(credentials + .cloudflare_refresh_token + .filter(|token| !token.trim().is_empty())) +} + +pub(crate) fn post_device_logout(api_base_url: &str, refresh_token: &str) -> Result<()> { + validate_credential_transport_url(api_base_url, "authentication API")?; + let url = format!("{}/api/devices/logout", api_base_url.trim_end_matches('/')); + match ureq::post(&url) + .timeout(AUTH_HTTP_TIMEOUT) + .send_json(serde_json::json!({ "refreshToken": refresh_token })) + { + Ok(_) => Ok(()), + Err(ureq::Error::Status(code, _)) => bail!("HTTP {code}"), + Err(error) => bail!("{error}"), + } +} + pub(crate) fn logout_backend( base: &Path, api_base_url: &str, diff --git a/crates/statsai/src/auth/tests.rs b/crates/statsai/src/auth/tests.rs index 29234ab..fe0fd82 100644 --- a/crates/statsai/src/auth/tests.rs +++ b/crates/statsai/src/auth/tests.rs @@ -1,4 +1,4 @@ -use super::logout::logout_backend; +use super::logout::{logout_backend, post_device_logout, revoke_server_session, ServerRevocation}; use super::session::{with_device_id_retry, DeviceSessionRequestError}; use super::*; use std::cell::Cell; @@ -192,6 +192,155 @@ fn logout_preserves_legacy_metadata_for_another_backend() { assert!(legacy_path.exists()); } +fn seed_logout_credentials( + base: &Path, + api_base_url: &str, + refresh_token: Option<&str>, +) -> PathBuf { + let path = auth_path_for_api_base_url(base, api_base_url); + let credentials = AuthCredentials { + backend: Some("cloudflare".to_string()), + api_base_url: Some(api_base_url.to_string()), + cloudflare_refresh_token: refresh_token.map(ToOwned::to_owned), + cloudflare_refresh_expires_at_secs: 10, + cloudflare_access_token: Some("access-token".to_string()), + cloudflare_access_expires_at_secs: 10, + device_id: Some("device-1".to_string()), + }; + write_credentials(&path, &credentials).expect("seed credentials"); + path +} + +/// Accepts one request, replies with `status_line`, and hands back the raw +/// request (headers and body) for inspection. +fn serve_one_logout_request( + status_line: &'static str, +) -> (String, std::sync::mpsc::Receiver) { + use std::io::Read; + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let address = listener.local_addr().expect("addr"); + let (sender, receiver) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept"); + let mut seen = Vec::new(); + let mut byte = [0_u8; 1]; + while stream.read(&mut byte).unwrap_or(0) == 1 { + seen.push(byte[0]); + if seen.ends_with(b"\r\n\r\n") { + break; + } + } + let length = String::from_utf8_lossy(&seen) + .lines() + .find_map(|line| { + line.to_ascii_lowercase() + .strip_prefix("content-length:") + .and_then(|value| value.trim().parse::().ok()) + }) + .unwrap_or(0); + let mut body = vec![0_u8; length]; + stream.read_exact(&mut body).expect("body"); + seen.extend_from_slice(&body); + let response = format!("{status_line}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + stream.write_all(response.as_bytes()).expect("respond"); + sender + .send(String::from_utf8_lossy(&seen).into_owned()) + .expect("send request"); + }); + (format!("http://{address}"), receiver) +} + +#[test] +fn logout_revokes_stored_refresh_token_on_the_server() { + let dir = tempfile::tempdir().expect("tempdir"); + let (api_base_url, request) = serve_one_logout_request("HTTP/1.1 204 No Content"); + seed_logout_credentials(dir.path(), &api_base_url, Some("refresh-secret")); + + let outcome = revoke_server_session(dir.path(), &api_base_url, post_device_logout); + + assert_eq!(outcome, ServerRevocation::Revoked); + let request = request + .recv_timeout(Duration::from_secs(5)) + .expect("request"); + assert!( + request.starts_with("POST /api/devices/logout HTTP/1.1\r\n"), + "{request}" + ); + let body = request.split("\r\n\r\n").nth(1).expect("body"); + let json: serde_json::Value = serde_json::from_str(body).expect("json body"); + assert_eq!( + json, + serde_json::json!({ "refreshToken": "refresh-secret" }) + ); +} + +#[test] +fn logout_reports_server_errors_and_still_clears_local_credentials() { + let dir = tempfile::tempdir().expect("tempdir"); + let (api_base_url, request) = serve_one_logout_request("HTTP/1.1 500 Internal Server Error"); + let path = seed_logout_credentials(dir.path(), &api_base_url, Some("refresh-secret")); + + let outcome = revoke_server_session(dir.path(), &api_base_url, post_device_logout); + + request + .recv_timeout(Duration::from_secs(5)) + .expect("request"); + match outcome { + ServerRevocation::Failed(reason) => assert!(reason.contains("HTTP 500"), "{reason}"), + other => panic!("expected a failed revocation, got {other:?}"), + } + assert!(path.exists(), "revocation must not touch local credentials"); + let removed = logout_backend(dir.path(), &api_base_url, |_| Ok(())).expect("logout"); + assert!(removed); + assert!(!path.exists()); +} + +#[test] +fn logout_reports_unreachable_server_without_failing() { + let dir = tempfile::tempdir().expect("tempdir"); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let api_base_url = format!("http://{}", listener.local_addr().expect("addr")); + drop(listener); + seed_logout_credentials(dir.path(), &api_base_url, Some("refresh-secret")); + + let outcome = revoke_server_session(dir.path(), &api_base_url, post_device_logout); + + assert!( + matches!(outcome, ServerRevocation::Failed(_)), + "{outcome:?}" + ); +} + +#[test] +fn logout_skips_the_server_without_a_stored_refresh_token() { + let dir = tempfile::tempdir().expect("tempdir"); + let api_base_url = "https://api.example.com"; + let calls = Cell::new(0); + let send = |_: &str, _: &str| { + calls.set(calls.get() + 1); + Ok(()) + }; + + assert_eq!( + revoke_server_session(dir.path(), api_base_url, send), + ServerRevocation::NotAttempted + ); + + seed_logout_credentials(dir.path(), api_base_url, None); + assert_eq!( + revoke_server_session(dir.path(), api_base_url, send), + ServerRevocation::NotAttempted + ); + assert_eq!(calls.get(), 0); +} + +#[test] +fn logout_revocation_refuses_plain_http_to_remote_hosts() { + let error = post_device_logout("http://api.example.com", "refresh-secret") + .expect_err("insecure transport"); + assert!(error.to_string().contains("must use HTTPS"), "{error:#}"); +} + #[test] fn atomic_auth_write_preserves_destination_on_serialization_failure() { struct FailingSerialize; diff --git a/crates/statsai/src/cli/args/mod.rs b/crates/statsai/src/cli/args/mod.rs index b472a16..97fa42b 100644 --- a/crates/statsai/src/cli/args/mod.rs +++ b/crates/statsai/src/cli/args/mod.rs @@ -128,8 +128,16 @@ pub(crate) enum AuthSubcommand { }, #[command(about = "Check authentication status for the Better Auth device session")] Status, - #[command(about = "Log out and clear stored Better Auth device credentials")] - Logout, + #[command( + about = "Log out: revoke this device's session on the server, then clear stored credentials" + )] + Logout { + #[arg( + long, + help = "Only clear credentials stored on this machine; do not contact the server" + )] + local_only: bool, + }, } #[derive(Debug, Args)] diff --git a/crates/statsai/src/cli/auth.rs b/crates/statsai/src/cli/auth.rs index b1e16ce..3028e4a 100644 --- a/crates/statsai/src/cli/auth.rs +++ b/crates/statsai/src/cli/auth.rs @@ -11,6 +11,6 @@ pub(crate) fn auth(command: AuthCommand) -> Result<()> { device_name, } => auth::login(no_open, headless, device_name), AuthSubcommand::Status => auth::status(), - AuthSubcommand::Logout => auth::logout(), + AuthSubcommand::Logout { local_only } => auth::logout(local_only), } } diff --git a/crates/statsai/tests/cli_surface/help.txt b/crates/statsai/tests/cli_surface/help.txt index 9183926..5ff45f7 100644 --- a/crates/statsai/tests/cli_surface/help.txt +++ b/crates/statsai/tests/cli_surface/help.txt @@ -1264,7 +1264,7 @@ Usage: statsai auth [OPTIONS] Commands: login Log in to the hosted sync backend status Check authentication status for the Better Auth device session - logout Log out and clear stored Better Auth device credentials + logout Log out: revoke this device's session on the server, then clear stored credentials help Print this message or the help of the given subcommand(s) Options: @@ -1293,11 +1293,12 @@ Options: --device-id Device identifier for multi-device sync -h, --help Print help ===== statsai auth logout --help ===== -Log out and clear stored Better Auth device credentials +Log out: revoke this device's session on the server, then clear stored credentials Usage: statsai auth logout [OPTIONS] Options: + --local-only Only clear credentials stored on this machine; do not contact the server --store Path to SQLite store --device-id Device identifier for multi-device sync -h, --help Print help From b30431a3eb881622f6d00be5945a7fab0cf7c2a9 Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 18:19:19 +0300 Subject: [PATCH 2/7] Show synced project paths under the home directory as ~ Sync payloads now rewrite a project path_label under the home directory to ~/... (the home directory itself to ~) for summaries, sessions, and task work items and spans. Only the outgoing label changes: the local store keeps the full path, path_hash and other identities are untouched, and pending-sync selection hashes the same payload that is sent. --- crates/statsai-core/src/paths.rs | 49 ++++ crates/statsai-core/src/tests/types.rs | 252 ++++++++++++++++++ crates/statsai-core/src/types/sync.rs | 88 +++++- crates/statsai-store/src/rollups/summaries.rs | 5 +- crates/statsai/src/cli/sync/batch.rs | 4 +- crates/statsai/src/cli/tests/sync/batch.rs | 63 +++++ docs/sync-contract.md | 11 +- 7 files changed, 458 insertions(+), 14 deletions(-) diff --git a/crates/statsai-core/src/paths.rs b/crates/statsai-core/src/paths.rs index f727d77..20ebd0e 100644 --- a/crates/statsai-core/src/paths.rs +++ b/crates/statsai-core/src/paths.rs @@ -46,6 +46,55 @@ pub fn home_dir() -> Option { .map(PathBuf::from) } +/// Rewrites a path label that names `home` or a path inside it as `~` or +/// `~/rest`, for labels that leave the device. Only the label text changes: +/// path hashes and other identities are computed from the full path and are +/// not affected. +/// +/// Matching is on whole path components, so `/Users/alice2` is not inside +/// `/Users/alice`. A Windows-style home (drive letter or backslash) matches +/// case-insensitively, accepts either separator, and keeps the label's own +/// separator after `~`. A home that is a filesystem or drive root never +/// matches, and labels outside home come back unchanged. +#[must_use] +pub fn collapse_home_path_label(label: &str, home: &Path) -> String { + let home = home.to_string_lossy(); + let windows = home.contains('\\') || has_windows_drive_prefix(&home); + let is_separator = |ch: char| ch == '/' || (windows && ch == '\\'); + let home = home.trim_end_matches(is_separator); + let is_root = home.is_empty() || (has_windows_drive_prefix(home) && home.len() == 2); + if is_root { + return label.to_string(); + } + let Some(prefix) = label.get(..home.len()) else { + return label.to_string(); + }; + let same_home = if windows { + prefix.bytes().zip(home.bytes()).all(|(left, right)| { + left.eq_ignore_ascii_case(&right) + || (matches!(left, b'/' | b'\\') && matches!(right, b'/' | b'\\')) + }) + } else { + prefix == home + }; + if !same_home { + return label.to_string(); + } + let rest = &label[home.len()..]; + if rest.chars().all(is_separator) { + return "~".to_string(); + } + if rest.starts_with(is_separator) { + return format!("~{rest}"); + } + label.to_string() +} + +fn has_windows_drive_prefix(value: &str) -> bool { + let bytes = value.as_bytes(); + bytes.len() >= 2 && bytes[0].is_ascii_alphabetic() && bytes[1] == b':' +} + #[must_use] pub fn expand_home_path(value: &str) -> PathBuf { if value == "~" { diff --git a/crates/statsai-core/src/tests/types.rs b/crates/statsai-core/src/tests/types.rs index e852347..c0a3f8c 100644 --- a/crates/statsai-core/src/tests/types.rs +++ b/crates/statsai-core/src/tests/types.rs @@ -913,3 +913,255 @@ fn assert_allowlist(value: &serde_json::Value, allowed: &[&str]) { ); } } + +#[test] +fn collapse_home_path_label_rewrites_paths_under_a_unix_home() { + let home = Path::new("/Users/alice"); + let cases = [ + ("/Users/alice/code/app", "~/code/app"), + ("/Users/alice", "~"), + ("/Users/alice/", "~"), + ("/Users/alice2/code/app", "/Users/alice2/code/app"), + ("/Users/alice2", "/Users/alice2"), + ("/Users/bob/code", "/Users/bob/code"), + ("/opt/work/app", "/opt/work/app"), + ("~/code/app", "~/code/app"), + ("/users/alice/code", "/users/alice/code"), + ("relative/Users/alice", "relative/Users/alice"), + ("", ""), + ]; + for (label, expected) in cases { + assert_eq!(collapse_home_path_label(label, home), expected, "{label}"); + } + assert_eq!( + collapse_home_path_label("/Users/alice/code", Path::new("/Users/alice/")), + "~/code" + ); +} + +#[test] +fn collapse_home_path_label_never_treats_a_root_home_as_home() { + assert_eq!( + collapse_home_path_label("/srv/app", Path::new("/")), + "/srv/app" + ); + assert_eq!( + collapse_home_path_label("/srv/app", Path::new("")), + "/srv/app" + ); + assert_eq!( + collapse_home_path_label(r"C:\work\app", Path::new(r"C:\")), + r"C:\work\app" + ); +} + +#[test] +fn collapse_home_path_label_handles_windows_style_homes() { + let home = Path::new(r"C:\Users\alice"); + let cases = [ + (r"C:\Users\alice\code\app", r"~\code\app"), + (r"C:\Users\alice", "~"), + (r"c:\users\ALICE\code", r"~\code"), + ("C:/Users/alice/code", "~/code"), + (r"C:\Users\alice2\code", r"C:\Users\alice2\code"), + (r"D:\Users\alice\code", r"D:\Users\alice\code"), + ]; + for (label, expected) in cases { + assert_eq!(collapse_home_path_label(label, home), expected, "{label}"); + } +} + +fn home_project(path_label: Option<&str>) -> ProjectInfo { + ProjectInfo { + project_id: "project_home".to_string(), + project_label: Some("app".to_string()), + repo_remote_hash: Some("remote-hash".to_string()), + repo_label: Some("owner/app".to_string()), + branch_hash: Some("branch-hash".to_string()), + branch_label: Some("main".to_string()), + path_hash: Some("path-hash".to_string()), + path_label: path_label.map(ToOwned::to_owned), + } +} + +#[test] +fn collapse_project_home_for_sync_only_rewrites_the_path_label() { + let home = Path::new("/Users/alice"); + let project = home_project(Some("/Users/alice/code/app")); + + let collapsed = collapse_project_home_for_sync(project.clone(), Some(home)); + + assert_eq!(collapsed.path_label.as_deref(), Some("~/code/app")); + assert_eq!( + ProjectInfo { + path_label: project.path_label.clone(), + ..collapsed + }, + project + ); + assert_eq!( + collapse_project_home_for_sync(project.clone(), None), + project + ); + assert_eq!( + collapse_project_home_for_sync(home_project(None), Some(home)), + home_project(None) + ); +} + +#[test] +fn sanitize_summary_for_sync_keeps_the_full_path_label() { + // Its output is also the locally stored payload; the home collapse is a + // separate step that only sync payloads take. + let now = mk_dt(2026, 5, 25); + let source = test_source("codex", "/tmp/codex"); + let mut summary = test_summary("codex", &source, now, now, now, 100); + summary.project = Some(home_project(Some("/Users/alice/code/app"))); + + let sanitized = sanitize_summary_for_sync(summary); + assert_eq!( + sanitized + .project + .as_ref() + .and_then(|project| project.path_label.as_deref()), + Some("/Users/alice/code/app") + ); + + let collapsed = collapse_summary_home_for_sync(sanitized, Some(Path::new("/Users/alice"))); + let project = collapsed.project.as_ref().expect("project"); + assert_eq!(project.path_label.as_deref(), Some("~/code/app")); + assert_eq!(project.path_hash.as_deref(), Some("path-hash")); + assert!(collapsed.privacy.contains_file_paths); +} + +#[test] +fn sanitize_session_rollup_for_sync_collapses_the_home_directory() { + let mut rollup = sample_session_rollup(mk_dt(2026, 5, 25)); + rollup.project = Some(home_project(Some("/Users/alice/code/app"))); + + let sanitized = + sanitize_session_rollup_for_sync_with_home(rollup.clone(), Some(Path::new("/Users/alice"))); + let project = sanitized.project.expect("project"); + assert_eq!(project.path_label.as_deref(), Some("~/code/app")); + assert_eq!(project.path_hash.as_deref(), Some("path-hash")); + + let outside = sanitize_session_rollup_for_sync_with_home(rollup, Some(Path::new("/Users/bob"))); + assert_eq!( + outside.project.and_then(|project| project.path_label), + Some("/Users/alice/code/app".to_string()) + ); +} + +fn home_task_span(path_label: &str) -> TaskSpan { + let now = mk_dt(2026, 5, 25); + TaskSpan { + schema_version: TASK_SPAN_SCHEMA_VERSION.to_string(), + span_id: TaskSpanId("span_home".to_string()), + provider: "codex".to_string(), + source_id: SourceId("source_home".to_string()), + span_kind: "test".to_string(), + source_record_id: Some("record".to_string()), + source_file_path_hash: None, + summary_id: None, + session_id: Some("session".to_string()), + thread_id: Some("thread".to_string()), + title: "Fix the build".to_string(), + normalized_title: "fix the build".to_string(), + title_source: None, + summary_preview: None, + todo_excerpt: None, + issue_keys: Vec::new(), + branch_family: None, + project_bucket: "bucket".to_string(), + project: Some(home_project(Some(path_label))), + git: None, + usage: UsageCounts::default(), + estimated_cost_usd: None, + estimated_cost_micro_usd: None, + event_count: 0, + has_usage_evidence: false, + total_messages: 0, + user_messages: 0, + assistant_messages: 0, + developer_messages: 0, + linked_event_ids: Vec::new(), + confidence: Confidence::Medium, + is_meta: false, + started_at: now, + ended_at: Some(now), + duration_seconds: Some(0), + } +} + +fn home_work_item(path_label: &str) -> WorkItem { + let now = mk_dt(2026, 5, 25); + WorkItem { + schema_version: WORK_ITEM_SCHEMA_VERSION.to_string(), + work_item_id: WorkItemId("work_home".to_string()), + anchor_span_id: TaskSpanId("span_home".to_string()), + tail_span_id: TaskSpanId("span_home".to_string()), + project_bucket: "bucket".to_string(), + title: "Fix the build".to_string(), + normalized_title: "fix the build".to_string(), + status: TaskStatus::Auto, + confidence: Confidence::Medium, + started_at: now, + ended_at: now, + duration_seconds: Some(0), + span_count: 1, + event_count: 0, + total_input_tokens: 0, + total_cache_creation_tokens: 0, + total_cache_read_tokens: 0, + total_output_tokens: 0, + total_reasoning_tokens: 0, + total_tokens: 0, + estimated_cost_usd: None, + estimated_cost_micro_usd: None, + providers: Vec::new(), + issue_keys: Vec::new(), + repo_label: None, + branch_labels: Vec::new(), + path_label: Some(path_label.to_string()), + summary_preview: None, + todo_excerpt: None, + no_git: true, + cross_provider: false, + continuation_reasons: Vec::new(), + review_reasons: Vec::new(), + } +} + +#[test] +fn sanitize_task_bucket_for_sync_collapses_work_item_and_span_paths() { + let snapshot = TaskBucketSnapshot { + project_bucket: "bucket".to_string(), + generated_at: mk_dt(2026, 5, 25), + applied_verification_cursor: None, + work_items: vec![ + home_work_item("/Users/alice/code/app"), + home_work_item("/Users/alice2/code/app"), + ], + members: Vec::new(), + spans: vec![home_task_span("/Users/alice")], + }; + + let sanitized = + sanitize_task_bucket_for_sync_with_home(snapshot, Some(Path::new("/Users/alice"))); + + assert_eq!( + sanitized.work_items[0].path_label.as_deref(), + Some("~/code/app") + ); + assert_eq!( + sanitized.work_items[1].path_label.as_deref(), + Some("/Users/alice2/code/app") + ); + let span = &sanitized.spans[0]; + let project = span.project.as_ref().expect("project"); + assert_eq!(project.path_label.as_deref(), Some("~")); + assert_eq!(project.path_hash.as_deref(), Some("path-hash")); + assert_eq!(span.source_record_id, None); + assert_eq!(span.session_id, None); + assert_eq!(span.thread_id, None); +} diff --git a/crates/statsai-core/src/types/sync.rs b/crates/statsai-core/src/types/sync.rs index d5bd18a..9d5aa49 100644 --- a/crates/statsai-core/src/types/sync.rs +++ b/crates/statsai-core/src/types/sync.rs @@ -1,14 +1,15 @@ use crate::{ - project_contains_file_paths, project_has_stable_identity, AccountEvidenceSummaryV1, - AccountPlanProjectionV1, ActivityCoverageV1, ActivityRollupV1, CodeChangeMetric, ProjectInfo, - ProviderAccount, ProviderAccountId, QuotaCycleContributionV1, SessionRollupV1, - SourceAccountAssignment, SourceAccountAssignmentId, SourceId, SourceLocation, Subscription, - SubscriptionId, SummaryId, TaskSpan, TaskVerification, TaskVerificationId, UsageEvent, - UsageSummary, WorkItem, WorkItemMember, + collapse_home_path_label, home_dir, project_contains_file_paths, project_has_stable_identity, + AccountEvidenceSummaryV1, AccountPlanProjectionV1, ActivityCoverageV1, ActivityRollupV1, + CodeChangeMetric, ProjectInfo, ProviderAccount, ProviderAccountId, QuotaCycleContributionV1, + SessionRollupV1, SourceAccountAssignment, SourceAccountAssignmentId, SourceId, SourceLocation, + Subscription, SubscriptionId, SummaryId, TaskSpan, TaskVerification, TaskVerificationId, + UsageEvent, UsageSummary, WorkItem, WorkItemMember, }; use chrono::{DateTime, Datelike, Duration, Utc}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use std::path::Path; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] pub struct SyncBatch { @@ -118,13 +119,33 @@ pub struct TaskBucketSnapshot { pub spans: Vec, } -/// Removes provider-local task locators before a snapshot leaves the device. +/// Removes provider-local task locators before a snapshot leaves the device, +/// and shows path labels under the home directory as `~/...`. #[must_use] -pub fn sanitize_task_bucket_for_sync(mut snapshot: TaskBucketSnapshot) -> TaskBucketSnapshot { +pub fn sanitize_task_bucket_for_sync(snapshot: TaskBucketSnapshot) -> TaskBucketSnapshot { + sanitize_task_bucket_for_sync_with_home(snapshot, home_dir().as_deref()) +} + +/// [`sanitize_task_bucket_for_sync`] with an explicit home directory. +#[must_use] +pub fn sanitize_task_bucket_for_sync_with_home( + mut snapshot: TaskBucketSnapshot, + home: Option<&Path>, +) -> TaskBucketSnapshot { + for item in &mut snapshot.work_items { + item.path_label = item + .path_label + .take() + .map(|label| collapse_home_label_for_sync(label, home)); + } for span in &mut snapshot.spans { span.source_record_id = None; span.session_id = None; span.thread_id = None; + span.project = span + .project + .take() + .map(|project| collapse_project_home_for_sync(project, home)); } snapshot } @@ -208,6 +229,43 @@ pub fn sanitize_project_for_sync(project: ProjectInfo) -> Option { Some(project) } +/// Shows a project path label under `home` as `~/...` before it leaves the +/// device. Only the label changes: `path_hash` and the other identity fields +/// keep the values computed from the full local path. +#[must_use] +pub fn collapse_project_home_for_sync( + mut project: ProjectInfo, + home: Option<&Path>, +) -> ProjectInfo { + project.path_label = project + .path_label + .take() + .map(|label| collapse_home_label_for_sync(label, home)); + project +} + +/// Applies [`collapse_project_home_for_sync`] to a summary's project. Kept out +/// of [`sanitize_summary_for_sync`], whose output is also the locally stored +/// payload and payload hash. +#[must_use] +pub fn collapse_summary_home_for_sync( + mut summary: UsageSummary, + home: Option<&Path>, +) -> UsageSummary { + summary.project = summary + .project + .take() + .map(|project| collapse_project_home_for_sync(project, home)); + summary +} + +fn collapse_home_label_for_sync(label: String, home: Option<&Path>) -> String { + match home { + Some(home) => collapse_home_path_label(&label, home), + None => label, + } +} + #[must_use] pub fn sanitize_summary_for_sync(mut summary: UsageSummary) -> UsageSummary { summary.source.source_record_id = None; @@ -242,12 +300,24 @@ const SYNC_FUTURE_SKEW: Duration = Duration::seconds(24 * 60 * 60); /// Values the ingest would reject are repaired here. One bad row fails the /// batch, and session chunks are sent before the snapshot, so a permanent /// rejection also leaves hosted rows that should have been retired. +/// +/// A project path label under the home directory is shown as `~/...`. +#[must_use] +pub fn sanitize_session_rollup_for_sync(rollup: SessionRollupV1) -> SessionRollupV1 { + sanitize_session_rollup_for_sync_with_home(rollup, home_dir().as_deref()) +} + +/// [`sanitize_session_rollup_for_sync`] with an explicit home directory. #[must_use] -pub fn sanitize_session_rollup_for_sync(mut rollup: SessionRollupV1) -> SessionRollupV1 { +pub fn sanitize_session_rollup_for_sync_with_home( + mut rollup: SessionRollupV1, + home: Option<&Path>, +) -> SessionRollupV1 { clamp_session_timestamps(&mut rollup); rollup.project = rollup .project .and_then(sanitize_project_for_sync) + .map(|project| collapse_project_home_for_sync(project, home)) .map(clamp_project_labels); clamp_session_title(&mut rollup); rollup.primary_model = rollup diff --git a/crates/statsai-store/src/rollups/summaries.rs b/crates/statsai-store/src/rollups/summaries.rs index 47b6b7d..252a31e 100644 --- a/crates/statsai-store/src/rollups/summaries.rs +++ b/crates/statsai-store/src/rollups/summaries.rs @@ -13,6 +13,9 @@ pub(crate) fn is_daily_rollup_summary(summary: &UsageSummary) -> bool { summary.metadata.summary_format == "daily_rollup.v1" } +/// The summary as an HTTP sync batch carries it: sanitized, without a project +/// unless projects are included, and with a project path label under the +/// home directory shown as `~/...`. Never stored locally. pub(crate) fn sanitize_summary_for_http_sync( summary: UsageSummary, include_projects: bool, @@ -21,7 +24,7 @@ pub(crate) fn sanitize_summary_for_http_sync( if !include_projects { summary.project = None; } - summary + statsai_core::collapse_summary_home_for_sync(summary, statsai_core::home_dir().as_deref()) } pub(crate) fn summary_sync_day(summary: &UsageSummary) -> NaiveDate { diff --git a/crates/statsai/src/cli/sync/batch.rs b/crates/statsai/src/cli/sync/batch.rs index 42e75e5..f6d1132 100644 --- a/crates/statsai/src/cli/sync/batch.rs +++ b/crates/statsai/src/cli/sync/batch.rs @@ -846,6 +846,8 @@ pub(crate) fn sanitize_summary_for_sync(summary: UsageSummary) -> UsageSummary { statsai_core::sanitize_summary_for_sync(summary) } +/// Matches the store's HTTP summary sanitizer, so pending-sync selection and +/// acknowledgements hash the same payload that is sent. pub(crate) fn sanitize_summary_for_sync_with_projects( summary: UsageSummary, include_projects: bool, @@ -854,7 +856,7 @@ pub(crate) fn sanitize_summary_for_sync_with_projects( if !include_projects { summary.project = None; } - summary + statsai_core::collapse_summary_home_for_sync(summary, statsai_core::home_dir().as_deref()) } pub(crate) fn is_daily_rollup_summary(summary: &UsageSummary) -> bool { diff --git a/crates/statsai/src/cli/tests/sync/batch.rs b/crates/statsai/src/cli/tests/sync/batch.rs index 0470ef5..1b2a8a3 100644 --- a/crates/statsai/src/cli/tests/sync/batch.rs +++ b/crates/statsai/src/cli/tests/sync/batch.rs @@ -2262,3 +2262,66 @@ fn receivers_advertise_the_cache_health_version_they_store() { )))); assert!(!remote_accepts_cache_health(None)); } + +#[test] +fn http_sync_shows_project_paths_under_home_as_tilde() { + let Some(home) = statsai_core::home_dir().filter(|home| home.parent().is_some()) else { + return; + }; + let full_path_label = home.join("work").join("app").to_string_lossy().into_owned(); + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "codex", + "test", + "0", + Path::new("/tmp/codex-http-home-label"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let mut event = test_event("codex", &source, Utc::now(), None, TokenParts::total(100)); + event.project = Some(ProjectInfo { + project_id: "project-home".to_string(), + project_label: Some("app".to_string()), + repo_remote_hash: None, + repo_label: None, + branch_hash: None, + branch_label: None, + path_hash: Some("path-hash".to_string()), + path_label: Some(full_path_label.clone()), + }); + store.insert_event(&event).expect("event"); + store.rebuild_sync_rollups().expect("rebuild"); + + let command = SyncCommand { + endpoint: Some("https://api.example.com/api/sync/batches".to_string()), + include_projects: true, + ..test_sync_command("http") + }; + let target = sync_target(&command).expect("target"); + let (batch, _) = build_sync_batch(&command, &store, "device", &target).expect("batch"); + + assert_eq!(batch.summaries.len(), 1); + let project = batch.summaries[0].project.as_ref().expect("project"); + let separator = std::path::MAIN_SEPARATOR; + assert_eq!( + project.path_label.as_deref(), + Some(format!("~{separator}work{separator}app").as_str()) + ); + assert_eq!(project.path_hash.as_deref(), Some("path-hash")); + let stored = store.all_sync_rollup_summaries().expect("stored rollups"); + assert_eq!( + stored[0] + .project + .as_ref() + .and_then(|project| project.path_label.as_deref()), + Some(full_path_label.as_str()), + "the local rollup keeps the full path" + ); + + record_rollup_sync_success(&store, "http", &target, &batch).expect("record sync"); + let (repeat, _) = build_sync_batch(&command, &store, "device", &target).expect("repeat"); + assert!( + repeat.summaries.is_empty(), + "an acknowledged rollup must not look pending again" + ); +} diff --git a/docs/sync-contract.md b/docs/sync-contract.md index 8dd6241..4ccfb18 100644 --- a/docs/sync-contract.md +++ b/docs/sync-contract.md @@ -239,9 +239,14 @@ through the branch's own commits. Conflict resolutions carried only by a merge commit are consequently not counted. `ProjectInfo.path_label` is retained for owner-facing project location displays, -manual project linking, and hosted task review. Hashed path, source, event, and -summary identifiers remain so the server can deduplicate records and keep -stable location identity. +manual project linking, and hosted task review. A path under the user's home +directory is sent as `~/...` (the home directory itself as `~`); paths outside +home are sent unchanged. This applies to `summaries[].project`, +`sessions[].project`, and task `work_items[].path_label` and +`spans[].project.path_label`. Only the label is rewritten: the local store keeps +the full path, and `path_hash` is still computed from it. Hashed path, source, +event, and summary identifiers remain so the server can deduplicate records and +keep stable location identity. Canonical provider account identity may now sync through `ProviderAccount.provider_user_id` and `ProviderAccount.email`. Hosted task From 4f1fc30c059e96aeadbab3ebe5035e027fd7bcd9 Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 18:20:30 +0300 Subject: [PATCH 3/7] Say what each opt-in sync flag sends The --include-projects, --include-tasks, --include-sessions, and --include-activity help now names the fields each one adds to hosted sync. The README lists the same, and the sync contract no longer says session rows carry no paths. --- README.md | 10 ++++++++++ crates/statsai/src/cli/args/sync.rs | 8 ++++---- crates/statsai/tests/cli_surface/help.txt | 8 ++++---- docs/sync-contract.md | 7 ++++--- 4 files changed, 22 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 36b8d1f..a988fab 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,16 @@ dashboard. | — | Opt-in private task snapshots and verifications, including bounded task titles, summary previews, and todo excerpts | | Session prompts, raw session ids, and message text | Opt-in session rollups: totals, duration, message counts, and a bounded title | +Each opt-in sync flag adds a specific set of fields to the hosted dataset: + +- `--include-projects`: project name, repository owner/repo, branch names, and + the project folder path, with the home directory shown as `~`. +- `--include-tasks`: task titles, short prompt previews, and todo excerpts. +- `--include-sessions`: per-session stats and a session title, which may be + derived from the first prompt. +- `--include-activity`: tool, MCP, plugin, skill, and command names with call + counts. + Raw usage events and complete archived conversation records stay local and are never included in hosted sync. StatsAI does not upload full prompts, full responses, or raw provider logs. When hosted task sync is explicitly enabled diff --git a/crates/statsai/src/cli/args/sync.rs b/crates/statsai/src/cli/args/sync.rs index 18a5b79..ace44da 100644 --- a/crates/statsai/src/cli/args/sync.rs +++ b/crates/statsai/src/cli/args/sync.rs @@ -50,7 +50,7 @@ pub(crate) struct SyncCommand { pub(crate) dry_run: bool, #[arg( long, - help = "Enable project metadata sync for this device and future syncs" + help = "Enable project metadata sync for this device and future syncs. Sends project name, repo owner/repo, branch names, and folder path (home shown as ~)" )] pub(crate) include_projects: bool, #[arg( @@ -62,7 +62,7 @@ pub(crate) struct SyncCommand { #[arg( long, conflicts_with_all = ["exclude_tasks", "exclude_projects"], - help = "Enable hosted task sync for this device and future syncs (implies --include-projects)" + help = "Enable hosted task sync for this device and future syncs (implies --include-projects). Sends task titles, short prompt previews, and todo excerpts" )] pub(crate) include_tasks: bool, #[arg( @@ -74,7 +74,7 @@ pub(crate) struct SyncCommand { #[arg( long, conflicts_with = "exclude_activity", - help = "Enable hosted agent activity sync for this device and future syncs" + help = "Enable hosted agent activity sync for this device and future syncs. Sends tool, MCP, plugin, skill, and command names with call counts" )] pub(crate) include_activity: bool, #[arg( @@ -86,7 +86,7 @@ pub(crate) struct SyncCommand { #[arg( long, conflicts_with_all = ["exclude_sessions", "exclude_projects"], - help = "Enable hosted session sync for this device and future syncs (implies --include-projects)" + help = "Enable hosted session sync for this device and future syncs (implies --include-projects). Sends per-session stats and a session title, which may be derived from the first prompt" )] pub(crate) include_sessions: bool, #[arg( diff --git a/crates/statsai/tests/cli_surface/help.txt b/crates/statsai/tests/cli_surface/help.txt index 5ff45f7..ee49870 100644 --- a/crates/statsai/tests/cli_surface/help.txt +++ b/crates/statsai/tests/cli_surface/help.txt @@ -1121,13 +1121,13 @@ Options: --reset-remote Delete mirrored hosted sync data for this paired device and clear local sync tracking (http only) --yes Confirm destructive sync reset actions --dry-run Build the sync batch without writing - --include-projects Enable project metadata sync for this device and future syncs + --include-projects Enable project metadata sync for this device and future syncs. Sends project name, repo owner/repo, branch names, and folder path (home shown as ~) --exclude-projects Disable project metadata sync for this device and future syncs; hosted sessions are retired - --include-tasks Enable hosted task sync for this device and future syncs (implies --include-projects) + --include-tasks Enable hosted task sync for this device and future syncs (implies --include-projects). Sends task titles, short prompt previews, and todo excerpts --exclude-tasks Disable hosted task sync for this device and future syncs - --include-activity Enable hosted agent activity sync for this device and future syncs + --include-activity Enable hosted agent activity sync for this device and future syncs. Sends tool, MCP, plugin, skill, and command names with call counts --exclude-activity Disable hosted agent activity sync and prune names on the next successful HTTP sync - --include-sessions Enable hosted session sync for this device and future syncs (implies --include-projects) + --include-sessions Enable hosted session sync for this device and future syncs (implies --include-projects). Sends per-session stats and a session title, which may be derived from the first prompt --exclude-sessions Disable hosted session sync for this device and future syncs -h, --help Print help ===== statsai schema --help ===== diff --git a/docs/sync-contract.md b/docs/sync-contract.md index 4ccfb18..67827fa 100644 --- a/docs/sync-contract.md +++ b/docs/sync-contract.md @@ -22,7 +22,7 @@ IDs. Versions 1–5 never treat absent activity fields as authoritative. Activit names leave the device only when the collector preference `include_activity` is on; turning it off sends an authoritative empty activity ID set so the backend prunes hosted names. Invocation IDs, arguments, outputs, commands, prompts, and -file paths never appear in a v6 payload. +file paths never appear in activity rows. The same batch schema carries opt-in `sessions` rows. Each row is `session_rollup.v1` with a closed key set: hashed session id, provider, source, account, time range, duration, usage, requests, cost, models, message @@ -33,8 +33,9 @@ sends it. `active_seconds` is the time the agent was working, the union of the turns the provider recorded, and is null when none were; collectors before it omit the key. Events keyed by their own provider record, such as Cursor's usage-export rows, belong to no session -and are not sent, and neither is a session with no project and no title. Prompts, responses, paths, and raw provider session ids stay on -the device. Session retirement is not implied by `sync_batch.v6`. A snapshot +and are not sent, and neither is a session with no project and no title. Prompts, responses, and raw provider session ids stay on +the device. The only path a session row carries is its project's `path_label`, +with the home directory shown as `~`. Session retirement is not implied by `sync_batch.v6`. A snapshot retires hosted sessions only when it includes `session_rollup_ids`. An empty array prunes them. Omitting the key leaves hosted sessions in place, which is what the collector does when `include_sessions` alone is off. With projects From e02bcb6fe321910a6a416dedaa450752336e7b5d Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 19:58:51 +0300 Subject: [PATCH 4/7] Re-send task buckets acknowledged under an older sync sanitizer Task buckets are re-sent only when they change, so buckets a target already held kept their full home paths after the sanitizer started showing them as ~. Each acknowledged bucket now records the sanitizer version it was sent under (schema 31), and incremental sync treats a bucket acknowledged under an older version as pending. Bumping TASK_BUCKET_SYNC_SANITIZER_VERSION re-sends them again for any future sanitizer change. Summaries and sessions already re-send because their payload hash changes; a test now covers that. --- crates/statsai-core/src/types/sync.rs | 8 ++ crates/statsai-store/src/migrations/mod.rs | 8 +- crates/statsai-store/src/migrations/v2.rs | 13 +++ .../src/pricing/tests/repricing.rs | 2 +- crates/statsai-store/src/sync_state/mod.rs | 19 ++-- crates/statsai-store/src/tasks/buckets.rs | 27 ++++-- crates/statsai-store/src/tasks/mod.rs | 3 +- .../statsai-store/src/tasks/tests/buckets.rs | 96 +++++++++++++++++++ crates/statsai/src/cli/tests/sync/batch.rs | 90 +++++++++++++++++ docs/sync-contract.md | 5 +- 10 files changed, 255 insertions(+), 16 deletions(-) diff --git a/crates/statsai-core/src/types/sync.rs b/crates/statsai-core/src/types/sync.rs index 9d5aa49..156714e 100644 --- a/crates/statsai-core/src/types/sync.rs +++ b/crates/statsai-core/src/types/sync.rs @@ -119,6 +119,14 @@ pub struct TaskBucketSnapshot { pub spans: Vec, } +/// Version of what [`sanitize_task_bucket_for_sync`] sends. Task buckets are +/// re-sent only when they change, so the collector records this version with +/// each acknowledged bucket and re-sends buckets acknowledged under an older +/// one. Bump it whenever the sanitizer changes what leaves the device. +/// +/// 1: path labels under the home directory are sent as `~/...`. +pub const TASK_BUCKET_SYNC_SANITIZER_VERSION: i64 = 1; + /// Removes provider-local task locators before a snapshot leaves the device, /// and shows path labels under the home directory as `~/...`. #[must_use] diff --git a/crates/statsai-store/src/migrations/mod.rs b/crates/statsai-store/src/migrations/mod.rs index a769d63..d0f5da1 100644 --- a/crates/statsai-store/src/migrations/mod.rs +++ b/crates/statsai-store/src/migrations/mod.rs @@ -8,7 +8,7 @@ mod v2; pub(crate) use v1::*; pub(crate) use v2::*; -pub const CURRENT_SCHEMA_VERSION: i64 = 30; +pub const CURRENT_SCHEMA_VERSION: i64 = 31; pub fn migrate(conn: &Connection) -> Result<()> { if let Some(current) = existing_schema_version(conn)? { @@ -155,6 +155,7 @@ fn apply_migration(conn: &Connection, version: i64) -> Result<()> { 28 => apply_migration_028(conn), 29 => apply_migration_029(conn), 30 => apply_migration_030(conn), + 31 => apply_migration_031(conn), _ => bail!("unsupported schema migration version {version}"), } } @@ -308,6 +309,11 @@ mod tests { ); assert!(sync_state_has_pending_resume_batch_id(&conn).expect("inspect sync_state")); assert!(table_exists(&conn, "task_bucket_sync_state")); + assert!(column_exists( + &conn, + "task_bucket_sync_state", + "sanitizer_version" + )); assert!(column_exists(&conn, "scan_file_state", "tasks_collected")); assert!(table_exists(&conn, "archive_missing_content_state")); assert!(table_exists(&conn, "archive_artifact_dependencies")); diff --git a/crates/statsai-store/src/migrations/v2.rs b/crates/statsai-store/src/migrations/v2.rs index 67c475c..d22de63 100644 --- a/crates/statsai-store/src/migrations/v2.rs +++ b/crates/statsai-store/src/migrations/v2.rs @@ -703,3 +703,16 @@ pub(crate) fn apply_migration_030(conn: &Connection) -> Result<()> { ))?; Ok(()) } + +/// Records which version of the task-bucket sync sanitizer produced each +/// acknowledged bucket. Rows acknowledged before this column existed read as +/// version 0, so a sanitizer change reaches buckets the target already holds +/// on the next incremental sync instead of waiting for the bucket to change. +pub(crate) fn apply_migration_031(conn: &Connection) -> Result<()> { + ensure_column( + conn, + "task_bucket_sync_state", + "sanitizer_version", + "INTEGER NOT NULL DEFAULT 0", + ) +} diff --git a/crates/statsai-store/src/pricing/tests/repricing.rs b/crates/statsai-store/src/pricing/tests/repricing.rs index a927256..3b31006 100644 --- a/crates/statsai-store/src/pricing/tests/repricing.rs +++ b/crates/statsai-store/src/pricing/tests/repricing.rs @@ -384,7 +384,7 @@ fn daily_rollups_are_unused_by_report_and_sync_paths() { assert_eq!(events[0].cost, expected_review_cost(started_at)); // Pin: bump this when CURRENT_SCHEMA_VERSION changes, after confirming // the daily_rollups table is still unused by report/sync/snapshot. - assert_eq!(CURRENT_SCHEMA_VERSION, 30); + assert_eq!(CURRENT_SCHEMA_VERSION, 31); } #[test] diff --git a/crates/statsai-store/src/sync_state/mod.rs b/crates/statsai-store/src/sync_state/mod.rs index 28018da..36540b3 100644 --- a/crates/statsai-store/src/sync_state/mod.rs +++ b/crates/statsai-store/src/sync_state/mod.rs @@ -702,21 +702,28 @@ impl Store { let local_buckets = self.task_project_buckets()?; let mut statement = self.conn.prepare( r#" - SELECT project_bucket, dirty + SELECT project_bucket, dirty, sanitizer_version FROM task_bucket_sync_state WHERE sink = ?1 AND target = ?2 AND device_id = ?3 "#, )?; let rows = statement.query_map(params![sink, target, device_id], |row| { - Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) + Ok(( + row.get::<_, String>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, i64>(2)?, + )) })?; - let mut tracked = HashMap::::new(); + let mut tracked = HashMap::::new(); for row in rows { - let (project_bucket, dirty) = row?; - tracked.insert(project_bucket, dirty); + let (project_bucket, dirty, sanitizer_version) = row?; + tracked.insert( + project_bucket, + !crate::tasks::task_bucket_ack_is_clean(dirty, sanitizer_version), + ); } let tracked_total = tracked.len() as u64; - let tracked_dirty = tracked.values().filter(|dirty| **dirty == 1).count() as u64; + let tracked_dirty = tracked.values().filter(|pending| **pending).count() as u64; let missing_local = local_buckets .iter() .filter(|project_bucket| !tracked.contains_key(project_bucket.as_str())) diff --git a/crates/statsai-store/src/tasks/buckets.rs b/crates/statsai-store/src/tasks/buckets.rs index be053f9..747050c 100644 --- a/crates/statsai-store/src/tasks/buckets.rs +++ b/crates/statsai-store/src/tasks/buckets.rs @@ -157,13 +157,15 @@ impl Store { let mut statement = self.conn.prepare( r#" INSERT INTO task_bucket_sync_state ( - sink, target, device_id, project_bucket, dirty, payload_hash, updated_at + sink, target, device_id, project_bucket, dirty, payload_hash, updated_at, + sanitizer_version ) - VALUES (?1, ?2, ?3, ?4, 0, ?5, ?6) + VALUES (?1, ?2, ?3, ?4, 0, ?5, ?6, ?7) ON CONFLICT(sink, target, device_id, project_bucket) DO UPDATE SET dirty = 0, payload_hash = excluded.payload_hash, - updated_at = excluded.updated_at + updated_at = excluded.updated_at, + sanitizer_version = excluded.sanitizer_version "#, )?; for snapshot in snapshots { @@ -174,6 +176,7 @@ impl Store { &snapshot.project_bucket, task_bucket_snapshot_payload_hash(snapshot)?, &now, + TASK_BUCKET_SYNC_SANITIZER_VERSION, ])?; } Ok(()) @@ -250,15 +253,19 @@ impl Store { self.conn .query_row( r#" - SELECT dirty + SELECT dirty, sanitizer_version FROM task_bucket_sync_state WHERE sink = ?1 AND target = ?2 AND device_id = ?3 AND project_bucket = ?4 "#, params![sink, target, device_id, project_bucket], - |row| row.get::<_, i64>(0), + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), ) .optional() - .map(|row| row.is_some_and(|dirty| dirty == 0)) + .map(|row| { + row.is_some_and(|(dirty, sanitizer_version)| { + task_bucket_ack_is_clean(dirty, sanitizer_version) + }) + }) .unwrap_or(false) } @@ -286,6 +293,14 @@ impl Store { } } +/// A bucket is clean for a target only when nothing changed since it was +/// acknowledged and the acknowledged payload went through the current sync +/// sanitizer. An older sanitizer version means the target holds a payload the +/// current one would not send, so the bucket goes out once more. +pub(crate) fn task_bucket_ack_is_clean(dirty: i64, sanitizer_version: i64) -> bool { + dirty == 0 && sanitizer_version >= TASK_BUCKET_SYNC_SANITIZER_VERSION +} + pub(crate) fn task_bucket_snapshot_payload_hash(snapshot: &TaskBucketSnapshot) -> Result { Ok(hash_text(&serde_json::to_string(snapshot)?)) } diff --git a/crates/statsai-store/src/tasks/mod.rs b/crates/statsai-store/src/tasks/mod.rs index 593a308..7caf0e6 100644 --- a/crates/statsai-store/src/tasks/mod.rs +++ b/crates/statsai-store/src/tasks/mod.rs @@ -5,7 +5,7 @@ use statsai_core::{ task_verification_id, title_topic_tokens, work_item_id, Confidence, TaskBucketSnapshot, TaskSpan, TaskSpanId, TaskStatus, TaskVerification, TaskVerificationAction, TaskVerificationCursor, UsageCounts, WorkItem, WorkItemId, WorkItemMember, - TASK_VERIFICATION_SCHEMA_VERSION, WORK_ITEM_SCHEMA_VERSION, + TASK_BUCKET_SYNC_SANITIZER_VERSION, TASK_VERIFICATION_SCHEMA_VERSION, WORK_ITEM_SCHEMA_VERSION, }; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::time::Instant; @@ -20,6 +20,7 @@ mod work_items; pub(crate) use super::{sqlite_in_clause_placeholders, sqlite_string_params}; pub(crate) use benchmark::*; +pub(crate) use buckets::task_bucket_ack_is_clean; pub(crate) use grouping::*; pub(crate) use titles::*; pub(crate) use verification::*; diff --git a/crates/statsai-store/src/tasks/tests/buckets.rs b/crates/statsai-store/src/tasks/tests/buckets.rs index ea7daa4..34ecd59 100644 --- a/crates/statsai-store/src/tasks/tests/buckets.rs +++ b/crates/statsai-store/src/tasks/tests/buckets.rs @@ -60,3 +60,99 @@ fn replacing_task_bucket_snapshot_marks_existing_sync_state_dirty() { "Implement sync dirty tracking v2" ); } + +#[test] +fn buckets_acknowledged_before_a_sanitizer_change_are_resent_once() { + let target = "https://example.invalid/api/sync/batches"; + let home = std::path::Path::new("/Users/alice"); + let path = tempfile::tempdir() + .expect("tempdir") + .keep() + .join("store.db"); + let mut snapshot = test_task_bucket_snapshot( + "bucket-home", + "span-home", + "Collapse home paths", + Utc.with_ymd_and_hms(2026, 7, 6, 10, 0, 0).unwrap(), + ); + snapshot.spans[0].project = Some(statsai_core::ProjectInfo { + project_id: "project-home".to_string(), + project_label: Some("app".to_string()), + repo_remote_hash: None, + repo_label: None, + branch_hash: None, + branch_label: None, + path_hash: Some("path-hash".to_string()), + path_label: Some("/Users/alice/code/app".to_string()), + }); + snapshot.work_items[0].path_label = Some("/Users/alice/code/app".to_string()); + + { + // A collector from before the sanitizer change syncs the bucket with + // its full path: its schema has no sanitizer version yet. + let store = Store::open(&path).expect("store"); + store + .replace_task_bucket_snapshot(&snapshot) + .expect("replace snapshot"); + store + .record_task_bucket_snapshots_synced( + "http", + target, + "device-1", + std::slice::from_ref(&snapshot), + ) + .expect("record legacy sync"); + store + .conn + .execute_batch( + "ALTER TABLE task_bucket_sync_state DROP COLUMN sanitizer_version; + DELETE FROM schema_migrations WHERE version = 31;", + ) + .expect("roll back to the previous schema"); + } + + // Upgrading reopens the store and migrates it. + let store = Store::open(&path).expect("reopened store"); + assert_eq!( + store + .task_bucket_sync_status("http", target, "device-1") + .expect("status") + .dirty, + 1 + ); + let pending = store + .pending_task_bucket_snapshots_for_sync("http", target, "device-1", false, None) + .expect("incremental selection after upgrade"); + assert_eq!( + pending.len(), + 1, + "the hosted bucket still has the full path" + ); + let sent = statsai_core::sanitize_task_bucket_for_sync_with_home( + pending.into_iter().next().expect("bucket"), + Some(home), + ); + assert_eq!(sent.work_items[0].path_label.as_deref(), Some("~/code/app")); + assert_eq!( + sent.spans[0] + .project + .as_ref() + .and_then(|project| project.path_label.as_deref()), + Some("~/code/app") + ); + + store + .record_task_bucket_snapshots_synced("http", target, "device-1", &[sent]) + .expect("record sync"); + assert!(store + .pending_task_bucket_snapshots_for_sync("http", target, "device-1", false, None) + .expect("incremental selection after resend") + .is_empty()); + assert_eq!( + store + .task_bucket_sync_status("http", target, "device-1") + .expect("status") + .dirty, + 0 + ); +} diff --git a/crates/statsai/src/cli/tests/sync/batch.rs b/crates/statsai/src/cli/tests/sync/batch.rs index 1b2a8a3..eb2b375 100644 --- a/crates/statsai/src/cli/tests/sync/batch.rs +++ b/crates/statsai/src/cli/tests/sync/batch.rs @@ -2325,3 +2325,93 @@ fn http_sync_shows_project_paths_under_home_as_tilde() { "an acknowledged rollup must not look pending again" ); } + +#[test] +fn http_sync_resends_rows_acknowledged_with_full_home_paths() { + let Some(home) = statsai_core::home_dir().filter(|home| home.parent().is_some()) else { + return; + }; + let full_path_label = home.join("work").join("app").to_string_lossy().into_owned(); + let separator = std::path::MAIN_SEPARATOR; + let collapsed_path_label = format!("~{separator}work{separator}app"); + let store = Store::in_memory().expect("store"); + let source = SourceLocation::local_adapter( + "codex", + "test", + "0", + Path::new("/tmp/codex-http-home-resend"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + let mut event = test_event("codex", &source, Utc::now(), None, TokenParts::total(100)); + event.session.session_id = "session-home".to_string(); + event.session.title = Some("Collapse home paths".to_string()); + event.project = Some(ProjectInfo { + project_id: "project-home".to_string(), + project_label: Some("app".to_string()), + repo_remote_hash: None, + repo_label: None, + branch_hash: None, + branch_label: None, + path_hash: Some("path-hash".to_string()), + path_label: Some(full_path_label.clone()), + }); + store.insert_event(&event).expect("event"); + store.rebuild_sync_rollups().expect("rebuild"); + + let command = SyncCommand { + endpoint: Some("https://api.example.com/api/sync/batches".to_string()), + include_projects: true, + include_sessions: true, + ..test_sync_command("http") + }; + let target = sync_target(&command).expect("target"); + let path_labels = |batch: &SyncBatch| { + let summaries = batch + .summaries + .iter() + .filter_map(|summary| summary.project.as_ref()?.path_label.clone()) + .collect::>(); + let sessions = batch + .sessions + .iter() + .filter_map(|session| session.project.as_ref()?.path_label.clone()) + .collect::>(); + (summaries, sessions) + }; + + // What a collector from before the change acknowledged: the same rows + // carrying the full path. + let (mut legacy, _) = build_sync_batch(&command, &store, "device", &target).expect("batch"); + assert_eq!(legacy.summaries.len(), 1); + assert_eq!(legacy.sessions.len(), 1); + for project in legacy + .summaries + .iter_mut() + .filter_map(|summary| summary.project.as_mut()) + .chain( + legacy + .sessions + .iter_mut() + .filter_map(|session| session.project.as_mut()), + ) + { + project.path_label = Some(full_path_label.clone()); + } + record_rollup_sync_success(&store, "http", &target, &legacy).expect("record legacy sync"); + + let (resend, _) = build_sync_batch(&command, &store, "device", &target).expect("resend"); + assert_eq!( + path_labels(&resend), + ( + vec![collapsed_path_label.clone()], + vec![collapsed_path_label.clone()] + ), + "rows acknowledged with the full path go out once more with ~" + ); + + record_rollup_sync_success(&store, "http", &target, &resend).expect("record resend"); + let (repeat, _) = build_sync_batch(&command, &store, "device", &target).expect("repeat"); + assert!(repeat.summaries.is_empty()); + assert!(repeat.sessions.is_empty()); +} diff --git a/docs/sync-contract.md b/docs/sync-contract.md index 67827fa..7f088c0 100644 --- a/docs/sync-contract.md +++ b/docs/sync-contract.md @@ -245,7 +245,10 @@ directory is sent as `~/...` (the home directory itself as `~`); paths outside home are sent unchanged. This applies to `summaries[].project`, `sessions[].project`, and task `work_items[].path_label` and `spans[].project.path_label`. Only the label is rewritten: the local store keeps -the full path, and `path_hash` is still computed from it. Hashed path, source, +the full path, and `path_hash` is still computed from it. After upgrading from +a collector that sent full paths, the next incremental sync re-sends affected +summaries and sessions once (their payload hash changed) and every task bucket +acknowledged under the older task sanitizer version. Hashed path, source, event, and summary identifiers remain so the server can deduplicate records and keep stable location identity. From cad647e8345729f4938ea688fafc05ed6147a81f Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 20:40:49 +0300 Subject: [PATCH 5/7] Hold the auth refresh lock through logout A sync refreshing at the same time could rotate the refresh token after logout read it. The server answers 204 for an already-revoked token, so logout reported success while the rotated session stayed active. Logout now takes the same lock token refresh uses before reading credentials and holds it through the server revocation and local cleanup, including with --local-only. --- crates/statsai/src/auth/logout.rs | 51 ++++++++++---- crates/statsai/src/auth/tests.rs | 109 +++++++++++++++++++++++++++++- 2 files changed, 147 insertions(+), 13 deletions(-) diff --git a/crates/statsai/src/auth/logout.rs b/crates/statsai/src/auth/logout.rs index a9452f2..3bc2b85 100644 --- a/crates/statsai/src/auth/logout.rs +++ b/crates/statsai/src/auth/logout.rs @@ -2,25 +2,52 @@ use super::*; pub fn logout(local_only: bool) -> Result<()> { let api_base_url = cloudflare_api_url(); - let base = auth_base_dir(); + let removed = logout_from_base( + &auth_base_dir(), + &api_base_url, + local_only, + post_device_logout, + delete_tokens_from_keyring_for_api_base_url, + )?; + if removed { + println!("Successfully logged out."); + } else { + println!("Already logged out."); + } + Ok(()) +} + +/// Revokes the server session (unless `local_only`) and clears local +/// credentials, all under the auth refresh lock. Holding it from before the +/// refresh token is read until the credentials are gone keeps a concurrent +/// sync from rotating the token in between: the server would accept the +/// revocation of the old token while the rotated session stayed active. +pub(crate) fn logout_from_base( + base: &Path, + api_base_url: &str, + local_only: bool, + send: impl FnOnce(&str, &str) -> Result<()>, + delete_keyring: impl FnOnce(&str) -> Result<()>, +) -> Result { + // With no auth directory there is nothing on disk to race over, and + // locking would create the directory just to log out. + let _refresh_lock = if base.exists() { + Some(acquire_auth_refresh_lock(&auth_path_for_api_base_url( + base, + api_base_url, + ))?) + } else { + None + }; if !local_only { - if let ServerRevocation::Failed(reason) = - revoke_server_session(&base, &api_base_url, post_device_logout) - { + if let ServerRevocation::Failed(reason) = revoke_server_session(base, api_base_url, send) { eprintln!("Warning: could not revoke this device's session on the server: {reason}"); eprintln!( "It expires within 30 days, or you can revoke it now from the Devices page of the dashboard." ); } } - if logout_backend(&base, &api_base_url, |api_base_url| { - delete_tokens_from_keyring_for_api_base_url(api_base_url) - })? { - println!("Successfully logged out."); - } else { - println!("Already logged out."); - } - Ok(()) + logout_backend(base, api_base_url, delete_keyring) } #[derive(Debug, PartialEq, Eq)] diff --git a/crates/statsai/src/auth/tests.rs b/crates/statsai/src/auth/tests.rs index fe0fd82..48c7e7d 100644 --- a/crates/statsai/src/auth/tests.rs +++ b/crates/statsai/src/auth/tests.rs @@ -1,4 +1,6 @@ -use super::logout::{logout_backend, post_device_logout, revoke_server_session, ServerRevocation}; +use super::logout::{ + logout_backend, logout_from_base, post_device_logout, revoke_server_session, ServerRevocation, +}; use super::session::{with_device_id_retry, DeviceSessionRequestError}; use super::*; use std::cell::Cell; @@ -334,6 +336,111 @@ fn logout_skips_the_server_without_a_stored_refresh_token() { assert_eq!(calls.get(), 0); } +fn auth_refresh_lock_is_held(base: &Path, api_base_url: &str) -> bool { + let lock_path = auth_path_for_api_base_url(base, api_base_url).with_extension("lock"); + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&lock_path) + .expect("open lock file"); + match file.try_lock() { + Ok(()) => false, + Err(std::fs::TryLockError::WouldBlock) => true, + Err(std::fs::TryLockError::Error(error)) => panic!("try lock: {error}"), + } +} + +#[test] +fn logout_holds_the_refresh_lock_through_revocation_and_cleanup() { + let dir = tempfile::tempdir().expect("tempdir"); + let api_base_url = "https://api.example.com"; + let path = seed_logout_credentials(dir.path(), api_base_url, Some("refresh-secret")); + let revoked = std::cell::RefCell::new(None); + + let removed = logout_from_base( + dir.path(), + api_base_url, + false, + |_, token| { + assert!(auth_refresh_lock_is_held(dir.path(), api_base_url)); + *revoked.borrow_mut() = Some(token.to_string()); + Ok(()) + }, + |_| { + assert!(auth_refresh_lock_is_held(dir.path(), api_base_url)); + Ok(()) + }, + ) + .expect("logout"); + + assert!(removed); + assert_eq!(revoked.borrow().as_deref(), Some("refresh-secret")); + assert!(!path.exists()); + assert!(!auth_refresh_lock_is_held(dir.path(), api_base_url)); +} + +#[test] +fn local_only_logout_clears_credentials_under_the_refresh_lock() { + let dir = tempfile::tempdir().expect("tempdir"); + let api_base_url = "https://api.example.com"; + let path = seed_logout_credentials(dir.path(), api_base_url, Some("refresh-secret")); + + let removed = logout_from_base( + dir.path(), + api_base_url, + true, + |_, _| panic!("--local-only must not contact the server"), + |_| { + assert!(auth_refresh_lock_is_held(dir.path(), api_base_url)); + Ok(()) + }, + ) + .expect("logout"); + + assert!(removed); + assert!(!path.exists()); +} + +#[test] +fn logout_waits_for_an_in_flight_refresh_and_revokes_the_rotated_token() { + let dir = tempfile::tempdir().expect("tempdir"); + let base = dir.path().to_path_buf(); + let api_base_url = "https://api.example.com"; + let path = seed_logout_credentials(&base, api_base_url, Some("refresh-old")); + + // A sync is mid-refresh: it holds the lock and is about to rotate. + let refresh_lock = acquire_auth_refresh_lock(&path).expect("refresh lock"); + let (sender, receiver) = std::sync::mpsc::channel(); + let logout_base = base.clone(); + let logout = std::thread::spawn(move || { + logout_from_base( + &logout_base, + api_base_url, + false, + |_, token| { + sender.send(token.to_string()).expect("send token"); + Ok(()) + }, + |_| Ok(()), + ) + }); + assert!( + receiver.recv_timeout(Duration::from_millis(200)).is_err(), + "logout must not read the token while a refresh holds the lock" + ); + seed_logout_credentials(&base, api_base_url, Some("refresh-rotated")); + drop(refresh_lock); + + assert_eq!( + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("revoked token"), + "refresh-rotated" + ); + assert!(logout.join().expect("logout thread").expect("logout")); + assert!(!path.exists()); +} + #[test] fn logout_revocation_refuses_plain_http_to_remote_hosts() { let error = post_device_logout("http://api.example.com", "refresh-secret") From 1509e1d56d0a3318c254d15489ba4db9f3f82d91 Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 21:03:00 +0300 Subject: [PATCH 6/7] Use one predicate for task bucket sync selection and status After the schema 31 migration, an acknowledged deletion tombstone read as sanitizer version 0. Sync status counted it as pending while sync selection, which only re-sends deleted buckets when dirty, never chose it, so task_bucket_backlog stayed above zero. A tombstone is an empty snapshot and no sanitizer change affects it, so only local buckets are re-sent for an older sanitizer version. Selection and status now share task_bucket_needs_sync and the same tracked-bucket query. --- crates/statsai-store/src/sync_state/mod.rs | 36 +---- crates/statsai-store/src/tasks/buckets.rs | 128 ++++++++++-------- crates/statsai-store/src/tasks/mod.rs | 1 - .../statsai-store/src/tasks/tests/buckets.rs | 90 ++++++++++++ 4 files changed, 167 insertions(+), 88 deletions(-) diff --git a/crates/statsai-store/src/sync_state/mod.rs b/crates/statsai-store/src/sync_state/mod.rs index 36540b3..2ff806b 100644 --- a/crates/statsai-store/src/sync_state/mod.rs +++ b/crates/statsai-store/src/sync_state/mod.rs @@ -365,7 +365,7 @@ impl Store { // rebuilds or deletes a bucket in that window updates nothing and leaves // no trace. Putting the captured `dirty = 0` back would then declare that // change already synced, and nothing compares hashes afterwards to catch - // it: `task_bucket_is_clean_for_sync` trusts this flag alone. Re-sending + // it: `task_bucket_needs_sync` trusts this flag, not hashes. Re-sending // the tracked buckets once costs a fraction of the re-upload restoring the // cursor avoids, and adoption made the same trade for the same reason. for (device_id, project_bucket, payload_hash, updated_at) in &snapshot.buckets { @@ -699,38 +699,10 @@ impl Store { target: &str, device_id: &str, ) -> Result { - let local_buckets = self.task_project_buckets()?; - let mut statement = self.conn.prepare( - r#" - SELECT project_bucket, dirty, sanitizer_version - FROM task_bucket_sync_state - WHERE sink = ?1 AND target = ?2 AND device_id = ?3 - "#, - )?; - let rows = statement.query_map(params![sink, target, device_id], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, i64>(2)?, - )) - })?; - let mut tracked = HashMap::::new(); - for row in rows { - let (project_bucket, dirty, sanitizer_version) = row?; - tracked.insert( - project_bucket, - !crate::tasks::task_bucket_ack_is_clean(dirty, sanitizer_version), - ); - } - let tracked_total = tracked.len() as u64; - let tracked_dirty = tracked.values().filter(|pending| **pending).count() as u64; - let missing_local = local_buckets - .iter() - .filter(|project_bucket| !tracked.contains_key(project_bucket.as_str())) - .count() as u64; + let selection = self.task_bucket_sync_selection(sink, target, device_id)?; Ok(TaskBucketSyncStatus { - total: tracked_total.saturating_add(missing_local), - dirty: tracked_dirty.saturating_add(missing_local), + total: selection.total as u64, + dirty: selection.pending.len() as u64, }) } diff --git a/crates/statsai-store/src/tasks/buckets.rs b/crates/statsai-store/src/tasks/buckets.rs index 747050c..1b017a1 100644 --- a/crates/statsai-store/src/tasks/buckets.rs +++ b/crates/statsai-store/src/tasks/buckets.rs @@ -83,23 +83,11 @@ impl Store { full: bool, applied_verification_cursor: Option, ) -> Result> { - let local_buckets = self.task_project_buckets()?; - let tracked_dirty_buckets = - self.dirty_task_bucket_keys_for_sync(sink, target, device_id)?; + let pending = self.task_bucket_sync_selection(sink, target, device_id)?; let bucket_ids = if full { - local_buckets - .union(&tracked_dirty_buckets) - .cloned() - .collect::>() + pending.local.union(&pending.pending).cloned().collect() } else { - local_buckets - .iter() - .filter(|bucket| { - !self.task_bucket_is_clean_for_sync(sink, target, device_id, bucket) - }) - .cloned() - .chain(tracked_dirty_buckets.difference(&local_buckets).cloned()) - .collect::>() + pending.pending }; bucket_ids .into_iter() @@ -219,54 +207,57 @@ impl Store { Ok(()) } - pub(crate) fn dirty_task_bucket_keys_for_sync( + /// Which task buckets the target still needs, by the one predicate both + /// sync selection and sync status use: [`task_bucket_needs_sync`]. + pub(crate) fn task_bucket_sync_selection( &self, sink: &str, target: &str, device_id: &str, - ) -> Result> { + ) -> Result { + let local = self.task_project_buckets()?; let mut statement = self.conn.prepare( r#" - SELECT project_bucket + SELECT project_bucket, dirty, sanitizer_version FROM task_bucket_sync_state - WHERE sink = ?1 AND target = ?2 AND device_id = ?3 AND dirty = 1 - ORDER BY project_bucket + WHERE sink = ?1 AND target = ?2 AND device_id = ?3 "#, )?; let rows = statement.query_map(params![sink, target, device_id], |row| { - row.get::<_, String>(0) + Ok(( + row.get::<_, String>(0)?, + TrackedTaskBucket { + dirty: row.get::<_, i64>(1)? != 0, + sanitizer_version: row.get::<_, i64>(2)?, + }, + )) })?; - let mut buckets = BTreeSet::new(); + let mut tracked = BTreeMap::new(); for row in rows { - buckets.insert(row?); + let (project_bucket, state) = row?; + tracked.insert(project_bucket, state); } - Ok(buckets) - } - - pub(crate) fn task_bucket_is_clean_for_sync( - &self, - sink: &str, - target: &str, - device_id: &str, - project_bucket: &str, - ) -> bool { - self.conn - .query_row( - r#" - SELECT dirty, sanitizer_version - FROM task_bucket_sync_state - WHERE sink = ?1 AND target = ?2 AND device_id = ?3 AND project_bucket = ?4 - "#, - params![sink, target, device_id, project_bucket], - |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), - ) - .optional() - .map(|row| { - row.is_some_and(|(dirty, sanitizer_version)| { - task_bucket_ack_is_clean(dirty, sanitizer_version) - }) + let pending = local + .iter() + .chain(tracked.keys()) + .filter(|bucket| { + task_bucket_needs_sync( + tracked.get(bucket.as_str()), + local.contains(bucket.as_str()), + ) }) - .unwrap_or(false) + .cloned() + .collect(); + let total = local + .iter() + .filter(|bucket| !tracked.contains_key(bucket.as_str())) + .count() + + tracked.len(); + Ok(TaskBucketSyncSelection { + local, + pending, + total, + }) } pub(crate) fn mark_task_buckets_dirty_in_tx( @@ -293,12 +284,39 @@ impl Store { } } -/// A bucket is clean for a target only when nothing changed since it was -/// acknowledged and the acknowledged payload went through the current sync -/// sanitizer. An older sanitizer version means the target holds a payload the -/// current one would not send, so the bucket goes out once more. -pub(crate) fn task_bucket_ack_is_clean(dirty: i64, sanitizer_version: i64) -> bool { - dirty == 0 && sanitizer_version >= TASK_BUCKET_SYNC_SANITIZER_VERSION +pub(crate) struct TaskBucketSyncSelection { + /// Buckets that exist locally. + pub(crate) local: BTreeSet, + /// Buckets an incremental sync sends: local or tracked ones the target + /// does not hold as they are now. + pub(crate) pending: BTreeSet, + /// Local buckets plus tracked ones that no longer exist locally. + pub(crate) total: usize, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct TrackedTaskBucket { + pub(crate) dirty: bool, + pub(crate) sanitizer_version: i64, +} + +/// Whether the target needs this bucket sent, given its acknowledgement (if +/// any) and whether the bucket still exists locally. +/// +/// A local bucket is pending until acknowledged, after any change, and when +/// it was acknowledged under an older sync sanitizer: the target then holds a +/// payload the current sanitizer would not send. A bucket that no longer +/// exists locally can only go out as an empty deletion tombstone, which +/// carries no labels, so only a change since its acknowledgement makes it +/// pending -- a sanitizer change alone would re-send a tombstone forever +/// without changing anything the target holds. +pub(crate) fn task_bucket_needs_sync(tracked: Option<&TrackedTaskBucket>, local: bool) -> bool { + match tracked { + None => local, + Some(state) => { + state.dirty || (local && state.sanitizer_version < TASK_BUCKET_SYNC_SANITIZER_VERSION) + } + } } pub(crate) fn task_bucket_snapshot_payload_hash(snapshot: &TaskBucketSnapshot) -> Result { diff --git a/crates/statsai-store/src/tasks/mod.rs b/crates/statsai-store/src/tasks/mod.rs index 7caf0e6..4d497f8 100644 --- a/crates/statsai-store/src/tasks/mod.rs +++ b/crates/statsai-store/src/tasks/mod.rs @@ -20,7 +20,6 @@ mod work_items; pub(crate) use super::{sqlite_in_clause_placeholders, sqlite_string_params}; pub(crate) use benchmark::*; -pub(crate) use buckets::task_bucket_ack_is_clean; pub(crate) use grouping::*; pub(crate) use titles::*; pub(crate) use verification::*; diff --git a/crates/statsai-store/src/tasks/tests/buckets.rs b/crates/statsai-store/src/tasks/tests/buckets.rs index 34ecd59..6f77b69 100644 --- a/crates/statsai-store/src/tasks/tests/buckets.rs +++ b/crates/statsai-store/src/tasks/tests/buckets.rs @@ -156,3 +156,93 @@ fn buckets_acknowledged_before_a_sanitizer_change_are_resent_once() { 0 ); } + +#[test] +fn acknowledged_tombstones_stay_clean_across_a_sanitizer_change() { + let target = "https://example.invalid/api/sync/batches"; + let path = tempfile::tempdir() + .expect("tempdir") + .keep() + .join("store.db"); + let started_at = Utc.with_ymd_and_hms(2026, 7, 6, 10, 0, 0).unwrap(); + let kept = test_task_bucket_snapshot("bucket-kept", "span-kept", "Keep me", started_at); + let gone = test_task_bucket_snapshot("bucket-gone", "span-gone", "Delete me", started_at); + + { + // A collector from before the sanitizer change syncs both buckets, + // deletes one locally, and syncs its tombstone. + let store = Store::open(&path).expect("store"); + for snapshot in [&kept, &gone] { + store + .replace_task_bucket_snapshot(snapshot) + .expect("replace snapshot"); + } + store + .record_task_bucket_snapshots_synced( + "http", + target, + "device-1", + &[kept.clone(), gone.clone()], + ) + .expect("record buckets"); + store + .replace_task_bucket_snapshot(&TaskBucketSnapshot { + project_bucket: "bucket-gone".to_string(), + generated_at: started_at, + applied_verification_cursor: None, + work_items: Vec::new(), + members: Vec::new(), + spans: Vec::new(), + }) + .expect("delete bucket locally"); + let tombstones = store + .pending_task_bucket_snapshots_for_sync("http", target, "device-1", false, None) + .expect("tombstone selection"); + assert_eq!(tombstones.len(), 1); + assert_eq!(tombstones[0].project_bucket, "bucket-gone"); + assert!(tombstones[0].spans.is_empty()); + store + .record_task_bucket_snapshots_synced("http", target, "device-1", &tombstones) + .expect("record tombstone"); + store + .conn + .execute_batch( + "ALTER TABLE task_bucket_sync_state DROP COLUMN sanitizer_version; + DELETE FROM schema_migrations WHERE version = 31;", + ) + .expect("roll back to the previous schema"); + } + + let store = Store::open(&path).expect("reopened store"); + let status = store + .task_bucket_sync_status("http", target, "device-1") + .expect("status"); + assert_eq!((status.dirty, status.total), (1, 2)); + let pending = store + .pending_task_bucket_snapshots_for_sync("http", target, "device-1", false, None) + .expect("incremental selection"); + assert_eq!( + pending + .iter() + .map(|snapshot| snapshot.project_bucket.as_str()) + .collect::>(), + vec!["bucket-kept"], + "only the bucket with content is re-sent" + ); + let full = store + .pending_task_bucket_snapshots_for_sync("http", target, "device-1", true, None) + .expect("full selection"); + assert_eq!(full.len(), 1); + + store + .record_task_bucket_snapshots_synced("http", target, "device-1", &pending) + .expect("record resend"); + let status = store + .task_bucket_sync_status("http", target, "device-1") + .expect("status after resend"); + assert_eq!((status.dirty, status.total), (0, 2)); + assert!(store + .pending_task_bucket_snapshots_for_sync("http", target, "device-1", false, None) + .expect("incremental selection after resend") + .is_empty()); +} From ce24e1e58537a7a5e6d75b085ba8118bdc64b856 Mon Sep 17 00:00:00 2001 From: Dmitry Starkov <21260939+starkdmi@users.noreply.github.com> Date: Thu, 8 Oct 2026 22:43:50 +0300 Subject: [PATCH 7/7] Finish an interrupted token rotation before revoking at logout If a refresh reached the server but its response was lost, the stored refresh token is already rotated away and its successor is live. The logout endpoint answers 204 for the old token without touching the successor, so logout reported success while a session stayed active. Under the refresh lock, logout now replays the pending rotation with its rotationId, as the next refresh would, and revokes the successor. If that exchange fails, logout warns that the session may still be active and how to revoke it, then clears local credentials. --- crates/statsai/src/auth/logout.rs | 31 +++++-- crates/statsai/src/auth/tests.rs | 135 ++++++++++++++++++++++++++++++ 2 files changed, 161 insertions(+), 5 deletions(-) diff --git a/crates/statsai/src/auth/logout.rs b/crates/statsai/src/auth/logout.rs index 3bc2b85..f49d802 100644 --- a/crates/statsai/src/auth/logout.rs +++ b/crates/statsai/src/auth/logout.rs @@ -41,9 +41,11 @@ pub(crate) fn logout_from_base( }; if !local_only { if let ServerRevocation::Failed(reason) = revoke_server_session(base, api_base_url, send) { - eprintln!("Warning: could not revoke this device's session on the server: {reason}"); eprintln!( - "It expires within 30 days, or you can revoke it now from the Devices page of the dashboard." + "Warning: could not revoke this device's session on the server, so it may still be active: {reason}" + ); + eprintln!( + "Revoke it from the Devices page of the dashboard; otherwise it expires within 30 days." ); } } @@ -60,12 +62,18 @@ pub(crate) enum ServerRevocation { /// Best-effort server-side revocation of the stored device session. Never /// fails: the caller clears local credentials whatever the outcome. +/// +/// The caller holds the auth refresh lock. If an earlier refresh left a +/// pending rotation -- the server may have rotated the token and the response +/// was lost -- the stored token may already be revoked while its successor is +/// live, and revoking the stored one would change nothing. The rotation is +/// replayed first, as the next refresh would, and the successor is revoked. pub(crate) fn revoke_server_session( base: &Path, api_base_url: &str, send: impl FnOnce(&str, &str) -> Result<()>, ) -> ServerRevocation { - let refresh_token = match stored_refresh_token(base, api_base_url) { + let refresh_token = match current_refresh_token(base, api_base_url) { Ok(Some(token)) => token, Ok(None) => return ServerRevocation::NotAttempted, Err(error) => return ServerRevocation::Failed(format!("{error:#}")), @@ -76,13 +84,26 @@ pub(crate) fn revoke_server_session( } } -fn stored_refresh_token(base: &Path, api_base_url: &str) -> Result> { +/// The refresh token that currently names this device's server session, +/// after finishing any interrupted rotation. +fn current_refresh_token(base: &Path, api_base_url: &str) -> Result> { if !auth_record_candidate_exists(base, api_base_url)? { return Ok(None); } - let Some((_path, credentials)) = auth_record_for_backend(base, api_base_url)? else { + let Some((path, mut credentials)) = auth_record_for_backend(base, api_base_url)? else { return Ok(None); }; + let has_token = credentials + .cloudflare_refresh_token + .as_deref() + .is_some_and(|token| !token.trim().is_empty()); + if !has_token { + return Ok(None); + } + if pending_refresh_rotation_path(&path).exists() { + refresh_cloudflare_access_token(&path, &mut credentials, api_base_url) + .context("finish an interrupted token refresh")?; + } Ok(credentials .cloudflare_refresh_token .filter(|token| !token.trim().is_empty())) diff --git a/crates/statsai/src/auth/tests.rs b/crates/statsai/src/auth/tests.rs index 48c7e7d..77af912 100644 --- a/crates/statsai/src/auth/tests.rs +++ b/crates/statsai/src/auth/tests.rs @@ -441,6 +441,141 @@ fn logout_waits_for_an_in_flight_refresh_and_revokes_the_rotated_token() { assert!(!path.exists()); } +/// Serves one request per response in order, closing each connection, and +/// hands back every raw request (headers and body). +fn serve_auth_requests( + responses: Vec<(&'static str, String)>, +) -> (String, std::sync::mpsc::Receiver) { + use std::io::Read; + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let address = listener.local_addr().expect("addr"); + let (sender, receiver) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + for (status_line, body) in responses { + let Ok((mut stream, _)) = listener.accept() else { + return; + }; + let mut seen = Vec::new(); + let mut byte = [0_u8; 1]; + while stream.read(&mut byte).unwrap_or(0) == 1 { + seen.push(byte[0]); + if seen.ends_with(b"\r\n\r\n") { + break; + } + } + let length = String::from_utf8_lossy(&seen) + .lines() + .find_map(|line| { + line.to_ascii_lowercase() + .strip_prefix("content-length:") + .and_then(|value| value.trim().parse::().ok()) + }) + .unwrap_or(0); + let mut request_body = vec![0_u8; length]; + stream.read_exact(&mut request_body).expect("body"); + seen.extend_from_slice(&request_body); + let response = format!( + "{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).expect("respond"); + if sender + .send(String::from_utf8_lossy(&seen).into_owned()) + .is_err() + { + return; + } + } + }); + (format!("http://{address}"), receiver) +} + +fn request_json(request: &str) -> serde_json::Value { + let body = request.split("\r\n\r\n").nth(1).expect("body"); + serde_json::from_str(body).expect("json body") +} + +#[test] +fn logout_finishes_an_interrupted_rotation_and_revokes_the_successor() { + let dir = tempfile::tempdir().expect("tempdir"); + let rotated = serde_json::json!({ + "accessToken": "access-new", + "accessExpiresAt": 4_102_444_800_u64, + "refreshToken": "refresh-new", + "refreshExpiresAt": 4_102_444_800_u64, + }) + .to_string(); + let (api_base_url, requests) = serve_auth_requests(vec![ + ("HTTP/1.1 200 OK", rotated), + ("HTTP/1.1 204 No Content", String::new()), + ]); + let path = seed_logout_credentials(dir.path(), &api_base_url, Some("refresh-old")); + // An earlier refresh reached the server, but its response was lost. + let rotation_id = + load_or_create_pending_refresh_rotation(&path, "refresh-old").expect("pending rotation"); + + let removed = logout_from_base(dir.path(), &api_base_url, false, post_device_logout, |_| { + Ok(()) + }) + .expect("logout"); + + let exchange = requests + .recv_timeout(Duration::from_secs(5)) + .expect("exchange"); + assert!( + exchange.starts_with("POST /api/devices/token HTTP/1.1\r\n"), + "{exchange}" + ); + let exchange = request_json(&exchange); + assert_eq!(exchange["refreshToken"], "refresh-old"); + assert_eq!(exchange["rotationId"], rotation_id.as_str()); + let revocation = requests + .recv_timeout(Duration::from_secs(5)) + .expect("revocation"); + assert!( + revocation.starts_with("POST /api/devices/logout HTTP/1.1\r\n"), + "{revocation}" + ); + assert_eq!( + request_json(&revocation), + serde_json::json!({ "refreshToken": "refresh-new" }) + ); + assert!(removed); + assert!(!path.exists()); + assert!(!pending_refresh_rotation_path(&path).exists()); +} + +#[test] +fn logout_reports_an_unfinished_rotation_and_still_clears_local_state() { + let dir = tempfile::tempdir().expect("tempdir"); + let (api_base_url, requests) = serve_auth_requests(vec![( + "HTTP/1.1 500 Internal Server Error", + "{}".to_string(), + )]); + let path = seed_logout_credentials(dir.path(), &api_base_url, Some("refresh-old")); + load_or_create_pending_refresh_rotation(&path, "refresh-old").expect("pending rotation"); + + let outcome = revoke_server_session(dir.path(), &api_base_url, |_, _| { + panic!("revoking the possibly rotated-away token would report a false success") + }); + match outcome { + ServerRevocation::Failed(reason) => { + assert!(reason.contains("interrupted token refresh"), "{reason}"); + assert!(reason.contains("HTTP 500"), "{reason}"); + } + other => panic!("expected a failed revocation, got {other:?}"), + } + assert!(requests + .recv_timeout(Duration::from_secs(5)) + .expect("exchange") + .starts_with("POST /api/devices/token HTTP/1.1\r\n")); + + let removed = logout_backend(dir.path(), &api_base_url, |_| Ok(())).expect("logout"); + assert!(removed); + assert!(!path.exists()); + assert!(!pending_refresh_rotation_path(&path).exists()); +} + #[test] fn logout_revocation_refuses_plain_http_to_remote_hosts() { let error = post_device_logout("http://api.example.com", "refresh-secret")