diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b004ed..eb088ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,12 @@ This changelog is automatically generated by AI when releases are published. For detailed design rationale and implementation notes, see `docs/superpowers/specs/2026-04-29-telemetry-streams-design.md`. +## [3.1.0] - 2026-11-08 + +### Fixed + +- **Repositories no org has connected stay on your computer**: in connected mode, authorship notes, commit summaries (with their prompts), file-change counts and prompt transcripts from a repository that none of your orgs has connected were uploaded to the org you signed in to — so a personal or client repository's prompts could show up in a workplace dashboard. Now each repository's data goes only to the org that connected it, and a repository no org has connected keeps its data on this computer, checked again once a day in case it gets connected. Transcripts now carry their repository, so they are routed the same way. Device-code logins (no personal access token) are routed too. `autter sync status` shows how many records are kept on this computer and why. See [changelog/unconnected-repos-stay-local.md](changelog/unconnected-repos-stay-local.md). + ## [3.0.0] - 2026-11-04 ### Added diff --git a/Cargo.lock b/Cargo.lock index be61d72..099b13a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -258,7 +258,7 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] name = "autter" -version = "3.0.0" +version = "3.1.0" dependencies = [ "autter", "base64", diff --git a/Cargo.toml b/Cargo.toml index 0126e3d..e1a68b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ resolver = "3" [package] name = "autter" -version = "3.0.0" +version = "3.1.0" edition = "2024" default-run = "autter" diff --git a/changelog/unconnected-repos-stay-local.md b/changelog/unconnected-repos-stay-local.md new file mode 100644 index 0000000..4ff33e7 --- /dev/null +++ b/changelog/unconnected-repos-stay-local.md @@ -0,0 +1,27 @@ +# Repositories no org has connected stay on your computer + +Before 3.1.0, connected mode sent every repository's data to an org. When autter could not tell which org a repository belonged to, it used the org you signed in to. That included repositories nobody had connected: a personal project or a client's repository worked on from a work login ended up in the workplace's dashboard, with its prompts, transcripts and file paths. + +## What changed + +- Before uploading, autter asks the server which of your orgs has connected the repository (by its default remote, `owner/repo`). The answer is cached for an hour. + - An org connected it: the data goes to that org. With a personal access token autter uses a token for that org; a device-code login uploads only when it is that login's org. + - No org connected it, or the commit has no remote: the data stays in the local queues and is not uploaded anywhere. It is offered again once a day, so connecting the repository on the web later uploads it. + - The server could not be reached: nothing is decided; the upload is retried on the normal schedule. +- Prompt transcripts now record their repository (`repo_url` in the transcript's metadata) and are routed like everything else. Transcripts queued by older versions have no repository, so they stay on this computer. +- The server checks the repository again on every upload and refuses data for a repository the token's org hasn't connected (`repo_not_connected`). autter keeps refused records locally instead of retrying them. +- Waiting for a repository to be connected is not counted as a failed upload, so it never makes autter give up on a record, and `autter sync status` doesn't report it as a stuck queue. + +## Checking it + +`autter sync status` lists records kept on this computer because their repository isn't connected: + +``` +Kept on this computer (repository not connected to any of your orgs): 4 authorship notes, 4 commit summaries, 3 transcripts, 12 file-change records +``` + +`autter sync status --json` reports the same numbers under `not_connected`. + +## Requires + +The updated Autter API: the repository lookup for device-code logins and the upload check ship with it. diff --git a/npm/package.json b/npm/package.json index 9b2fef1..d720717 100644 --- a/npm/package.json +++ b/npm/package.json @@ -1,6 +1,6 @@ { "name": "@autter/cli", - "version": "3.0.0", + "version": "3.1.0", "description": "Git-native AI authorship tracking: records which lines were written by AI agents and links them to the agent, model, and prompt. This package is a thin bootstrapper that installs the native autter binary.", "license": "Apache-2.0", "homepage": "https://autter.dev", diff --git a/src/api/client.rs b/src/api/client.rs index 6ccdeb6..863ac71 100644 --- a/src/api/client.rs +++ b/src/api/client.rs @@ -286,8 +286,14 @@ fn acquire_refresh_file_lock() -> Option { /// Prefix marking a Personal Access Token (mirrors the backend). const PAT_PREFIX: &str = "autter_pat_"; -/// In-process cache: repo remote URL → resolved owning org id (None = untracked). -static ORG_RESOLVE_CACHE: Lazy>>> = +/// How long a repo → org answer is trusted before asking the server again, so a +/// repo connected (or disconnected) on the web is picked up by a running daemon. +const ORG_RESOLVE_TTL_SECS: i64 = 3_600; + +/// In-process cache: repo remote URL → (owning org id, None = no org connected it; +/// unix time it was resolved). Only answers from the server are cached, never +/// errors: an outage must not turn into "this repo isn't connected". +static ORG_RESOLVE_CACHE: Lazy, i64)>>> = Lazy::new(|| Mutex::new(std::collections::HashMap::new())); /// In-process cache: org id → (access token, unix expiry). @@ -295,8 +301,8 @@ static ORG_TOKEN_CACHE: Lazy Option { let creds = CredentialStore::new().load().ok().flatten()?; if creds.is_refresh_token_expired() { @@ -308,29 +314,108 @@ fn load_stored_pat() -> Option { Some(creds.refresh_token) } -/// Resolve which org owns the repo at `repo_url`, cached in-process. Returns None -/// when no org tracks it, the user isn't logged in with a PAT, or resolution fails -/// — callers then fall back to the home org. -pub fn resolve_org_for_repo_cached(repo_url: &str) -> Option { +/// Error text a queue stores on items parked because their repo isn't connected +/// to any of the user's orgs. Matches the server's rejection reason. +pub const REPO_NOT_CONNECTED: &str = "repo_not_connected"; + +/// How long parked items wait before their repo is checked again. +pub const UNCONNECTED_RECHECK_SECS: i64 = 86_400; + +/// Ask the server which of the user's orgs connected `repo_url`. Uses the PAT +/// when logged in with one, otherwise the device login's access token. +/// `Ok(None)`: no org connected it. `Err`: couldn't find out right now. +fn resolve_org_for_repo( + repo_url: &str, + access_token: Option<&str>, +) -> Result, String> { + let now = chrono::Utc::now().timestamp(); if let Ok(cache) = ORG_RESOLVE_CACHE.lock() - && let Some(hit) = cache.get(repo_url) + && let Some((hit, at)) = cache.get(repo_url) + && now - *at < ORG_RESOLVE_TTL_SECS { - return hit.clone(); + return Ok(hit.clone()); } - let pat = load_stored_pat()?; - let resolved = OAuthClient::new() - .resolve_org_for_repo(&pat, repo_url) - .ok() - .flatten(); + let client = OAuthClient::new(); + let resolved = if let Some(pat) = load_stored_pat() { + client.resolve_org_for_repo(&pat, repo_url)? + } else if let Some(token) = access_token { + client.resolve_org_for_repo_with_token(token, repo_url)? + } else { + return Err("not logged in".to_string()); + }; if let Ok(mut cache) = ORG_RESOLVE_CACHE.lock() { - cache.insert(repo_url.to_string(), resolved.clone()); + cache.insert(repo_url.to_string(), (resolved.clone(), now)); + } + Ok(resolved) +} + +/// Resolve which org owns the repo at `repo_url`, cached in-process. Returns None +/// when no org has connected it, or it can't be resolved right now. For reads +/// (memory, badge, wrapped); uploads use [`route_for_repo`], which tells those +/// two cases apart. +pub fn resolve_org_for_repo_cached(repo_url: &str) -> Option { + let token = try_load_auth_token().ok(); + resolve_org_for_repo(repo_url, token.as_deref()).ok().flatten() +} + +/// Where one repo's data may be uploaded. +pub enum RepoRoute { + /// Upload with this client; its token is scoped to the org that connected the repo. + Upload(ApiClient), + /// None of the user's orgs connected this repo (or the login can't reach the + /// org that did). Keep the data on the machine and check again later + /// ([`UNCONNECTED_RECHECK_SECS`]); never send it to the login org instead. + NotConnected, + /// Couldn't decide right now (offline, not logged in, server error). Retry + /// on the normal schedule. + Unavailable(String), +} + +/// Decide where data for `repo_url` goes. Data without a repo is kept local: +/// there is no way to show it belongs to any org. +pub fn route_for_repo(repo_url: Option<&str>) -> RepoRoute { + route_for_repo_at(repo_url, None) +} + +/// [`route_for_repo`] against a specific API base URL (`None` = configured default). +pub fn route_for_repo_at(repo_url: Option<&str>, base_url: Option) -> RepoRoute { + let Some(repo_url) = repo_url.map(str::trim).filter(|url| !url.is_empty()) else { + return RepoRoute::NotConnected; + }; + let default = ApiClient::new(ApiContext::new(base_url.clone())); + let Some(login_token) = default.context().auth_token.clone() else { + // An API key is bound to one org and can't be routed; the server still + // refuses repos that org hasn't connected. + if default.has_api_key() { + return RepoRoute::Upload(default); + } + return RepoRoute::Unavailable("not logged in".to_string()); + }; + let org = match resolve_org_for_repo(repo_url, Some(&login_token)) { + Ok(Some(org)) => org, + Ok(None) => return RepoRoute::NotConnected, + Err(err) => return RepoRoute::Unavailable(err), + }; + let login_org = crate::auth::identity::extract_identity_from_access_token(&login_token) + .active_org_id; + if login_org.as_deref() == Some(org.as_str()) { + return RepoRoute::Upload(default); + } + if load_stored_pat().is_none() { + // A device login is scoped to one org and can't mint a token for + // another. The repo belongs to that other org, not the login org. + return RepoRoute::NotConnected; + } + match access_token_for_org(&org) { + Some(token) => RepoRoute::Upload(ApiClient::new(ApiContext::with_auth(base_url, token))), + None => RepoRoute::Unavailable(format!("could not get a token for org {}", org)), } - resolved } /// Get an access token scoped to `org_id`, minted from the stored PAT and cached -/// in-process until near expiry. Returns None if not logged in with a PAT or the -/// mint fails (caller should NOT fall back to another org — retry instead). +/// in-process until near expiry. Without a PAT, returns the login token only if +/// it is already scoped to `org_id`. Returns None otherwise or if the mint fails +/// (caller should NOT fall back to another org — retry instead). pub fn access_token_for_org(org_id: &str) -> Option { let now = chrono::Utc::now().timestamp(); if let Ok(cache) = ORG_TOKEN_CACHE.lock() @@ -339,7 +424,13 @@ pub fn access_token_for_org(org_id: &str) -> Option { { return Some(token.clone()); } - let pat = load_stored_pat()?; + let Some(pat) = load_stored_pat() else { + // A device login can't mint tokens, but its own token already works + // for the org it signed in to. + let token = try_load_auth_token().ok()?; + let identity = crate::auth::identity::extract_identity_from_access_token(&token); + return (identity.active_org_id.as_deref() == Some(org_id)).then_some(token); + }; let creds = OAuthClient::new() .exchange_pat_for_org(&pat, Some(org_id)) .ok()?; diff --git a/src/api/data_plane.rs b/src/api/data_plane.rs index 0bb14a8..bccd66b 100644 --- a/src/api/data_plane.rs +++ b/src/api/data_plane.rs @@ -27,6 +27,13 @@ fn require_success( Ok(body.to_string()) } +/// True when an upload failed because the server refused the repo: it isn't +/// connected to the token's org. The data should be kept local, not retried soon. +pub fn is_repo_not_connected(err: &AutterError) -> bool { + let message = err.to_string(); + message.contains("status 403") && message.contains(crate::api::client::REPO_NOT_CONNECTED) +} + impl ApiClient { pub fn upload_commit_authorship_summary( &self, diff --git a/src/api/types.rs b/src/api/types.rs index 7cc3ae2..83f4b41 100644 --- a/src/api/types.rs +++ b/src/api/types.rs @@ -141,6 +141,17 @@ pub struct NotesUploadRequest { pub struct NotesUploadResponse { pub success_count: usize, pub failure_count: usize, + /// Notes the server refused because their repo isn't connected to the + /// token's org. Absent from older servers. + #[serde(default)] + pub rejected: Vec, +} + +/// One note the server refused, and why (`repo_not_connected`). +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct NotesUploadRejection { + pub commit_sha: String, + pub error: String, } /// Response from a notes read request — maps commit_sha → note content. @@ -211,6 +222,10 @@ pub struct FileChangeUploadFailure { pub struct FileChangeUploadResponse { #[serde(default)] pub failed: Vec, + /// Rows the server refused because their repo isn't connected to the + /// token's org. Absent from older servers. + #[serde(default)] + pub rejected: Vec, } /// Single result from CA prompt store batch read diff --git a/src/auth/client.rs b/src/auth/client.rs index 5dbbaf7..52bd00d 100644 --- a/src/auth/client.rs +++ b/src/auth/client.rs @@ -292,15 +292,34 @@ impl OAuthClient { .map_err(|e| format!("Token sign-in failed: {}", e)) } - /// Resolve which org owns a repository (by `owner/repo` or remote URL), so a - /// push can be routed to the right org. Returns `None` when no org tracks it - /// (the caller should fall back to the PAT's home org). + /// Resolve which of the PAT owner's orgs has connected a repository (by + /// `owner/repo` or remote URL), so a push can be routed to the right org. + /// `Ok(None)` means no org has connected it: its data stays on the machine. pub fn resolve_org_for_repo(&self, pat: &str, repo: &str) -> Result, String> { - let url = format!("{}/worker/oauth/resolve-org", self.base_url); - let body = serde_json::json!({ "pat": pat, "repo": repo }); + self.post_resolve_org(serde_json::json!({ "pat": pat, "repo": repo }), None) + } + + /// Same as [`Self::resolve_org_for_repo`] for a device-code login, which has + /// no PAT: the caller is identified by its access token instead. + pub fn resolve_org_for_repo_with_token( + &self, + access_token: &str, + repo: &str, + ) -> Result, String> { + self.post_resolve_org(serde_json::json!({ "repo": repo }), Some(access_token)) + } + fn post_resolve_org( + &self, + body: serde_json::Value, + bearer: Option<&str>, + ) -> Result, String> { + let url = format!("{}/worker/oauth/resolve-org", self.base_url); let (_agent, request) = ApiContext::http_post(&url, Some(30)); - let request = request.set("Content-Type", "application/json"); + let mut request = request.set("Content-Type", "application/json"); + if let Some(token) = bearer { + request = request.set("Authorization", &format!("Bearer {}", token)); + } let response = http::send_with_body(request, &body.to_string()) .map_err(|e| format!("Failed to connect to server: {}", e))?; diff --git a/src/auth/notice.rs b/src/auth/notice.rs index 1034700..e0a7769 100644 --- a/src/auth/notice.rs +++ b/src/auth/notice.rs @@ -536,12 +536,15 @@ fn read_pending_sync_counts() -> Option { let file_changes = crate::file_changes::FileChangesDatabase::global() .ok() .and_then(|db| db.lock().ok()?.count_pending().ok())?; + // Items parked because their repo isn't connected aren't waiting on the + // upload pipeline; counting them would make a healthy sync look stuck. + let kept = crate::upload_hold::unconnected_counts(); Some(PendingSyncCounts { metrics, - notes, - commit_summaries, - transcripts, - file_changes, + notes: (notes - kept.notes).max(0), + commit_summaries: (commit_summaries - kept.commit_summaries).max(0), + transcripts: (transcripts - kept.transcripts).max(0), + file_changes: (file_changes - kept.file_changes).max(0), }) } diff --git a/src/authorship/cas_bridge.rs b/src/authorship/cas_bridge.rs index 7f7d9ed..cb884cb 100644 --- a/src/authorship/cas_bridge.rs +++ b/src/authorship/cas_bridge.rs @@ -49,11 +49,14 @@ pub enum TranscriptDestination { /// /// `checkpoints` is the working log for the commit being processed; it is the /// only place the transcript file path is available (via `agent_metadata`). +/// `repo_url` is the commit's remote: it travels with each transcript so the +/// upload only goes to an org that connected that repo. pub fn enqueue_prompt_transcripts( prompts: &mut BTreeMap, sessions: &mut BTreeMap, checkpoints: &[Checkpoint], destination: TranscriptDestination, + repo_url: Option<&str>, ) { if prompts.is_empty() && sessions.is_empty() { return; @@ -71,7 +74,9 @@ pub fn enqueue_prompt_transcripts( let Some(path) = transcript_paths.get(hash) else { continue; }; - if let Ok(Some(cas_hash)) = enqueue_transcript_file(path, &record.agent_id, destination) { + if let Ok(Some(cas_hash)) = + enqueue_transcript_file(path, &record.agent_id, destination, repo_url) + { record.messages_url = Some(format!("cas:{cas_hash}")); } } @@ -83,7 +88,9 @@ pub fn enqueue_prompt_transcripts( let Some(path) = transcript_paths.get(session_key) else { continue; }; - if let Ok(Some(cas_hash)) = enqueue_transcript_file(path, &record.agent_id, destination) { + if let Ok(Some(cas_hash)) = + enqueue_transcript_file(path, &record.agent_id, destination, repo_url) + { record.messages_url = Some(format!("cas:{cas_hash}")); } } @@ -117,6 +124,7 @@ fn enqueue_transcript_file( path: &str, agent_id: &AgentId, destination: TranscriptDestination, + repo_url: Option<&str>, ) -> Result, AutterError> { let Some(messages) = messages_from_transcript_file(path)? else { return Ok(None); @@ -141,6 +149,9 @@ fn enqueue_transcript_file( metadata.insert("tool".to_string(), agent_id.tool.clone()); metadata.insert("model".to_string(), agent_id.model.clone()); metadata.insert("session_id".to_string(), agent_id.id.clone()); + if let Some(repo_url) = repo_url { + metadata.insert("repo_url".to_string(), repo_url.to_string()); + } let db = InternalDatabase::global()?; let mut db_lock = db diff --git a/src/authorship/internal_db.rs b/src/authorship/internal_db.rs index a546885..160ac5b 100644 --- a/src/authorship/internal_db.rs +++ b/src/authorship/internal_db.rs @@ -528,6 +528,43 @@ impl InternalDatabase { Ok(self.conn.execute(&sql, params.as_slice())?) } + /// Park transcripts whose repo isn't connected to any of the user's orgs. + /// They stay queued and are offered again after + /// [`crate::api::client::UNCONNECTED_RECHECK_SECS`]. `attempts` is left + /// alone, so waiting for a repo to be connected never gives up on them. + pub fn defer_cas_unconnected(&mut self, hashes: &[String]) -> Result { + if hashes.is_empty() { + return Ok(0); + } + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let placeholders: Vec<&str> = hashes.iter().map(|_| "?").collect(); + let sql = format!( + "UPDATE cas_sync_queue + SET status = 'pending', last_sync_error = ?, last_sync_at = ?, + next_retry_at = ?, processing_started_at = NULL + WHERE hash IN ({})", + placeholders.join(",") + ); + let error = crate::api::client::REPO_NOT_CONNECTED; + let next_retry_at = now.saturating_add(crate::api::client::UNCONNECTED_RECHECK_SECS); + let mut params: Vec<&dyn rusqlite::ToSql> = vec![&error, &now, &next_retry_at]; + params.extend(hashes.iter().map(|hash| hash as &dyn rusqlite::ToSql)); + Ok(self.conn.execute(&sql, params.as_slice())?) + } + + /// Transcripts kept on this machine because their repo isn't connected to + /// any of the user's orgs. + pub fn count_unconnected_cas(&self) -> Result { + Ok(self.conn.query_row( + "SELECT COUNT(*) FROM cas_sync_queue WHERE last_sync_error = ?1", + params![crate::api::client::REPO_NOT_CONNECTED], + |row| row.get(0), + )?) + } + /// Get cached CAS messages by hash pub fn get_cas_cache(&self, hash: &str) -> Result, AutterError> { let result = self.conn.query_row( diff --git a/src/authorship/post_commit.rs b/src/authorship/post_commit.rs index b721840..e2d9061 100644 --- a/src/authorship/post_commit.rs +++ b/src/authorship/post_commit.rs @@ -220,6 +220,7 @@ pub fn post_commit_with_final_state( &mut authorship_log.metadata.sessions, &parent_working_log, destination, + current_repo_url(repo).as_deref(), ); } @@ -646,9 +647,7 @@ fn upload_commit_authorship_summary( checkpoints: &[Checkpoint], hunks_json: Option<&str>, ) { - use crate::api::client::{ - ApiClient, ApiContext, access_token_for_org, resolve_org_for_repo_cached, - }; + use crate::api::client::{RepoRoute, route_for_repo}; use crate::api::types::CommitAuthorshipSummary; // The summary queue is upload-only (nothing local reads it), so local mode @@ -718,21 +717,21 @@ fn upload_commit_authorship_summary( // Attempt an immediate upload for the normal online case. Every early // return below is safe because the durable daemon queue owns the retry. - let client = if let Some(repo_url) = row.repo_url.as_deref() - && let Some(org) = resolve_org_for_repo_cached(repo_url) - { - let Some(token) = access_token_for_org(&org) else { + // A repo that no org of the user's has connected stays on this machine: + // never fall back to the login org, which would put a personal or client + // repo's prompts into a workplace's dashboard. + let client = match route_for_repo(row.repo_url.as_deref()) { + RepoRoute::Upload(client) => client, + RepoRoute::NotConnected => { + if let Ok(db) = crate::notes::db::NotesDatabase::global() { + let mut lock = db.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let _ = lock.defer_unconnected_commit_summary(commit_sha); + } return; - }; - ApiClient::new(ApiContext::with_auth(None, token)) - } else { - ApiClient::new(ApiContext::new(None)) + } + RepoRoute::Unavailable(_) => return, }; - if !client.is_logged_in() { - return; - } - match client.upload_commit_authorship_summary(&row, &crate::config::get_or_create_distinct_id()) { Ok(()) => { diff --git a/src/commands/sync.rs b/src/commands/sync.rs index 5b66e1b..3ec1959 100644 --- a/src/commands/sync.rs +++ b/src/commands/sync.rs @@ -14,9 +14,33 @@ use std::io::IsTerminal; #[derive(Debug, Serialize)] struct SyncStatusOutput { cloud_sync: crate::auth::notice::CloudSyncStatusReport, + /// Records kept on this machine because their repository isn't connected + /// to any of your orgs. They are never uploaded to another org. + not_connected: NotConnectedCounts, ok: bool, } +#[derive(Debug, Serialize)] +struct NotConnectedCounts { + notes: i64, + commit_summaries: i64, + transcripts: i64, + file_changes: i64, + total: i64, +} + +impl From for NotConnectedCounts { + fn from(counts: crate::upload_hold::QueueCounts) -> Self { + Self { + notes: counts.notes, + commit_summaries: counts.commit_summaries, + transcripts: counts.transcripts, + file_changes: counts.file_changes, + total: counts.total(), + } + } +} + pub fn handle_sync(args: &[String]) { match args.first().map(|s| s.as_str()) { None => print_status(args), @@ -176,6 +200,7 @@ fn print_status(args: &[String]) { if arg_parser::json() { let output = SyncStatusOutput { cloud_sync: report, + not_connected: crate::upload_hold::unconnected_counts().into(), ok, }; println!( @@ -220,6 +245,17 @@ fn print_human_status(report: &crate::auth::notice::CloudSyncStatusReport) { println!("Pending uploads: none"); } + let kept = crate::upload_hold::unconnected_counts(); + if kept.total() > 0 { + println!( + "Kept on this computer (repository not connected to any of your orgs): {} authorship notes, {} commit summaries, {} transcripts, {} file-change records", + kept.notes, kept.commit_summaries, kept.transcripts, kept.file_changes + ); + println!( + " They upload once an org connects the repository (checked daily). They are never sent to another org." + ); + } + if let Some(remediation) = &report.remediation { println!(); println!("Fix: {remediation}"); diff --git a/src/daemon/telemetry_worker.rs b/src/daemon/telemetry_worker.rs index af1d3d0..f2e5e62 100644 --- a/src/daemon/telemetry_worker.rs +++ b/src/daemon/telemetry_worker.rs @@ -433,24 +433,19 @@ fn flush_commit_summaries() { } }; - let client = if let Some(org) = row - .repo_url - .as_deref() - .and_then(crate::api::client::resolve_org_for_repo_cached) - { - match crate::api::client::access_token_for_org(&org) { - Some(token) => ApiClient::new(ApiContext::with_auth(None, token)), - None => { - let mut lock = db.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); - let _ = lock.mark_commit_summary_failed( - &pending_row.commit_sha, - "could not mint org-scoped token", - ); - continue; - } + // Only upload to an org that connected this repo; otherwise keep it here. + let client = match crate::api::client::route_for_repo(row.repo_url.as_deref()) { + crate::api::client::RepoRoute::Upload(client) => client, + crate::api::client::RepoRoute::NotConnected => { + let mut lock = db.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let _ = lock.defer_unconnected_commit_summary(&pending_row.commit_sha); + continue; + } + crate::api::client::RepoRoute::Unavailable(reason) => { + let mut lock = db.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let _ = lock.mark_commit_summary_failed(&pending_row.commit_sha, &reason); + continue; } - } else { - ApiClient::new(ApiContext::new(None)) }; let result = client.upload_commit_authorship_summary(&row, &get_or_create_distinct_id()); @@ -464,6 +459,10 @@ fn flush_commit_summaries() { crate::auth::notice::clear_upload_failure_for("commit_summaries"); } } + Err(error) if crate::api::data_plane::is_repo_not_connected(&error) => { + // The server knows better than our cached answer: park it. + let _ = lock.defer_unconnected_commit_summary(&pending_row.commit_sha); + } Err(error) => { let _ = lock.mark_commit_summary_failed(&pending_row.commit_sha, &error.to_string()); @@ -1106,24 +1105,21 @@ pub fn flush_notes() { return; } - // Route each note to the org that owns its repo. Notes whose repo isn't known - // or isn't tracked by any org go to the home org (org = None → default token). + // Route each note to the org that connected its repo. A note whose repo no + // org of the user's has connected (or that has no repo) stays on this + // machine; it is never sent to the login org instead. type NoteBatchRow = (String, String, Option); let mut groups: std::collections::HashMap, Vec> = std::collections::HashMap::new(); for note in &pending { - let org = note - .repo_url - .as_deref() - .and_then(crate::api::client::resolve_org_for_repo_cached); - groups.entry(org).or_default().push(( + groups.entry(note.repo_url.clone()).or_default().push(( note.commit_sha.clone(), note.content.clone(), note.repo_url.clone(), )); } - for (org_opt, batch) in groups { + for (repo_opt, batch) in groups { let commit_shas: Vec = batch.iter().map(|(sha, _, _)| sha.clone()).collect(); let entries: Vec = batch .iter() @@ -1134,65 +1130,75 @@ pub fn flush_notes() { }) .collect(); - // Pick the client for this org. A resolved org mints an org-scoped token; - // if minting fails we defer (mark failed → retry) rather than misroute. - let group_client = match &org_opt { - Some(org) => match crate::api::client::access_token_for_org(org) { - Some(token) => { - ApiClient::new(ApiContext::with_auth(Some(backend_url.clone()), token)) + let group_client = match crate::api::client::route_for_repo_at( + repo_opt.as_deref(), + Some(backend_url.clone()), + ) { + crate::api::client::RepoRoute::Upload(client) => client, + crate::api::client::RepoRoute::NotConnected => { + if let Ok(db) = crate::notes::db::NotesDatabase::global() + && let Ok(mut lock) = db.lock() + { + let _ = lock.defer_unconnected(&commit_shas); } - None => { - if let Ok(db) = crate::notes::db::NotesDatabase::global() - && let Ok(mut lock) = db.lock() - { - let _ = lock.mark_failed(&commit_shas, "could not mint org-scoped token"); - } - continue; + continue; + } + crate::api::client::RepoRoute::Unavailable(reason) => { + if let Ok(db) = crate::notes::db::NotesDatabase::global() + && let Ok(mut lock) = db.lock() + { + let _ = lock.mark_failed(&commit_shas, &reason); } - }, - None => ApiClient::new(ApiContext::new(Some(backend_url.clone()))), + continue; + } }; let request = NotesUploadRequest { entries }; match group_client.upload_notes(request) { Ok(resp) => { + // Notes the server refused because the repo isn't connected are + // parked; only the rest count as uploaded or failed. + let rejected: std::collections::HashSet<&str> = resp + .rejected + .iter() + .filter(|r| r.error == crate::api::client::REPO_NOT_CONNECTED) + .map(|r| r.commit_sha.as_str()) + .collect(); + let (parked, rest): (Vec, Vec) = commit_shas + .iter() + .cloned() + .partition(|sha| rejected.contains(sha.as_str())); + let other_failures = resp.failure_count.saturating_sub(parked.len()); + if resp.success_count > 0 { note_durable_sync_authenticated(); - if resp.failure_count == 0 { + if other_failures == 0 { crate::auth::notice::clear_upload_failure_for("notes"); } } tracing::debug!( success = resp.success_count, failure = resp.failure_count, - org = org_opt.as_deref().unwrap_or("home"), + not_connected = parked.len(), "notes: uploaded batch" ); if let Ok(db) = crate::notes::db::NotesDatabase::global() && let Ok(mut lock) = db.lock() { - if resp.failure_count == 0 { - let _ = lock.mark_synced(&commit_shas); + let _ = lock.defer_unconnected(&parked); + if other_failures == 0 { + let _ = lock.mark_synced(&rest); } else { // Server reported partial failures but doesn't identify which - // entries failed. Mark the whole group failed so all entries + // entries failed. Mark the rest of the group failed so they // retry on the next flush cycle. - note_durable_sync_upload_failed( - "notes", - &format!( - "partial failure: {}/{} entries failed", - resp.failure_count, - commit_shas.len() - ), - ); - let _ = lock.mark_failed( - &commit_shas, - &format!( - "partial failure: {}/{} entries failed", - resp.failure_count, - commit_shas.len() - ), + let reason = format!( + "partial failure: {}/{} entries failed", + other_failures, + rest.len() ); + note_durable_sync_upload_failed("notes", &reason); + let _ = lock.mark_failed(&rest, &reason); } } } @@ -1229,20 +1235,27 @@ pub fn flush_notes() { /// via `Config::notes_backend_url()`); otherwise fall back to the API base /// URL (legacy behavior). fn cas_client() -> (ApiClient, bool) { + let (client, enabled, _) = cas_target(); + (client, enabled) +} + +/// [`cas_client`], plus the data-plane base URL when it is the hosted one +/// (`None` for a self-hosted plane, which has a single org and no routing). +fn cas_target() -> (ApiClient, bool, Option>) { let cfg = Config::fresh(); let dataplane_url = if cfg.notes_backend_kind().uses_http() { cfg.notes_backend_url().map(|s| s.to_string()) } else { None }; - let context = ApiContext::new(dataplane_url); + let context = ApiContext::new(dataplane_url.clone()); let target_url = context.base_url.clone(); let client = ApiClient::new(context); let using_hosted = target_url == crate::config::DEFAULT_API_BASE_URL || target_url == crate::config::DEFAULT_NOTES_BACKEND_URL; let enabled = !using_hosted || client.is_logged_in() || client.has_api_key(); - (client, enabled) + (client, enabled, using_hosted.then_some(dataplane_url)) } fn flush_cas(records: Vec) { @@ -1254,7 +1267,7 @@ fn flush_cas(records: Vec) { ); return; } - let (client, enabled) = cas_client(); + let (client, enabled, hosted_base) = cas_target(); if !enabled { tracing::debug!("telemetry: skipping CAS flush, not logged in"); return; @@ -1288,6 +1301,46 @@ fn flush_cas(records: Vec) { return; } + // On the hosted plane, each transcript goes to the org that connected its + // repo (from `metadata.repo_url`). Transcripts whose repo no org of the + // user's has connected, or with no repo at all, stay on this machine. + let Some(base_url) = hosted_base else { + upload_cas_objects(&client, cas_objects); + return; + }; + let mut by_repo: std::collections::HashMap, Vec> = + std::collections::HashMap::new(); + for object in cas_objects { + by_repo + .entry(object.metadata.get("repo_url").cloned()) + .or_default() + .push(object); + } + for (repo_url, objects) in by_repo { + match crate::api::client::route_for_repo_at(repo_url.as_deref(), base_url.clone()) { + crate::api::client::RepoRoute::Upload(repo_client) => { + upload_cas_objects(&repo_client, objects); + } + crate::api::client::RepoRoute::NotConnected => { + let hashes: Vec = objects.iter().map(|o| o.hash.clone()).collect(); + defer_cas_unconnected(&hashes); + } + crate::api::client::RepoRoute::Unavailable(reason) => { + let hashes: Vec = objects.iter().map(|o| o.hash.clone()).collect(); + mark_cas_upload_failed(&hashes, &reason); + } + } + } +} + +fn defer_cas_unconnected(hashes: &[String]) { + if let Ok(db) = crate::authorship::internal_db::InternalDatabase::global() { + let mut lock = db.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let _ = lock.defer_cas_unconnected(hashes); + } +} + +fn upload_cas_objects(client: &ApiClient, cas_objects: Vec) { for chunk in cas_objects.chunks(50) { let hashes: Vec = chunk.iter().map(|o| o.hash.clone()).collect(); let request = CasUploadRequest { @@ -1311,12 +1364,17 @@ fn flush_cas(records: Vec) { note_durable_sync_authenticated(); crate::auth::notice::clear_upload_failure_for("transcripts"); } - let failed_hashes: Vec = response - .results - .iter() - .filter(|result| result.status != "ok") - .map(|result| result.hash.clone()) - .collect(); + // Refused because the repo isn't connected: park, don't retry. + let mut parked_hashes = Vec::new(); + let mut failed_hashes = Vec::new(); + for result in response.results.iter().filter(|r| r.status != "ok") { + if result.error.as_deref() == Some(crate::api::client::REPO_NOT_CONNECTED) { + parked_hashes.push(result.hash.clone()); + } else { + failed_hashes.push(result.hash.clone()); + } + } + defer_cas_unconnected(&parked_hashes); if !failed_hashes.is_empty() { mark_cas_upload_failed(&failed_hashes, "org database rejected CAS object"); } diff --git a/src/file_changes/db.rs b/src/file_changes/db.rs index 9dacbdd..30b5aac 100644 --- a/src/file_changes/db.rs +++ b/src/file_changes/db.rs @@ -283,6 +283,16 @@ impl FileChangesDatabase { Ok(count) } + /// Rows kept on this machine because their repo isn't connected to any of + /// the user's orgs. + pub fn count_unconnected(&self) -> Result { + Ok(self.conn.query_row( + "SELECT COUNT(*) FROM file_change_counts WHERE synced = 0 AND last_sync_error = ?1", + params![crate::api::client::REPO_NOT_CONNECTED], + |row| row.get(0), + )?) + } + /// Hold every row waiting for upload. Local counts are unaffected; the /// rows are just never dequeued until [`Self::release_held`]. A held row /// touched again stays held, because its aggregate includes held data. diff --git a/src/file_changes/mod.rs b/src/file_changes/mod.rs index ce522ec..e9af7de 100644 --- a/src/file_changes/mod.rs +++ b/src/file_changes/mod.rs @@ -61,7 +61,9 @@ pub fn top_changed_files(repo_key: &str, limit: usize) -> Result, Vec> = + // Each repo's counts go only to the org that connected that repo. A repo no + // org of the user's has connected keeps its counts on this machine. + let mut groups: std::collections::HashMap> = std::collections::HashMap::new(); for row in pending { - let org = resolve_org_for_repo_cached(&row.repo_url); - groups.entry(org).or_default().push(row); + groups.entry(row.repo_url.clone()).or_default().push(row); } - for (org_opt, batch) in groups { - let client = match &org_opt { - Some(org) => match access_token_for_org(org) { - Some(token) => ApiClient::new(crate::api::client::ApiContext::with_auth( - Some(backend_url.clone()), - token, - )), - None => { - mark_batch_failed(&batch, "could not mint org-scoped token", 300); - continue; - } - }, - None => ApiClient::new(crate::api::client::ApiContext::new(Some( - backend_url.clone(), - ))), + for (repo_url, batch) in groups { + let client = match route_for_repo_at(Some(&repo_url), Some(backend_url.clone())) { + RepoRoute::Upload(client) => client, + RepoRoute::NotConnected => { + mark_batch_failed(&batch, REPO_NOT_CONNECTED, UNCONNECTED_RECHECK_SECS as u64); + continue; + } + RepoRoute::Unavailable(reason) => { + mark_batch_failed(&batch, &reason, 300); + continue; + } }; let rows: Vec = batch @@ -168,16 +167,21 @@ pub fn flush_pending_to_cloud() { .collect(); match client.upload_file_change_counts(&rows, &distinct_id) { - Ok(FileChangeUploadResponse { failed }) => { + Ok(FileChangeUploadResponse { failed, rejected }) => { let failed_keys: std::collections::HashSet<(String, String)> = failed .into_iter() .map(|row| (row.repo_url, row.file_path)) .collect(); + let rejected_keys: std::collections::HashSet<(String, String)> = rejected + .into_iter() + .map(|row| (row.repo_url, row.file_path)) + .collect(); let mut by_repo: std::collections::HashMap> = std::collections::HashMap::new(); for row in &batch { - if failed_keys.contains(&(row.repo_url.clone(), row.file_path.clone())) { + let key = (row.repo_url.clone(), row.file_path.clone()); + if failed_keys.contains(&key) || rejected_keys.contains(&key) { continue; } by_repo @@ -201,6 +205,21 @@ pub fn flush_pending_to_cloud() { } } + // Refused by the server because the repo isn't connected: park. + if !rejected_keys.is_empty() + && let Ok(db) = FileChangesDatabase::global() + && let Ok(mut lock) = db.lock() + { + for (repo_url, file_path) in rejected_keys { + let _ = lock.mark_failed( + &repo_url, + &[file_path], + REPO_NOT_CONNECTED, + UNCONNECTED_RECHECK_SECS as u64, + ); + } + } + if !failed_keys.is_empty() { crate::auth::notice::record_upload_failure( "file_changes", diff --git a/src/notes/db.rs b/src/notes/db.rs index faf80cb..d297c53 100644 --- a/src/notes/db.rs +++ b/src/notes/db.rs @@ -562,6 +562,53 @@ impl NotesDatabase { Ok(()) } + /// Park notes whose repo isn't connected to any of the user's orgs. They + /// stay on disk and are offered again after + /// [`crate::api::client::UNCONNECTED_RECHECK_SECS`], in case the repo gets + /// connected. `attempts` is left alone: this isn't a failure, and counting + /// it would eventually give up on the note for good. + pub fn defer_unconnected(&mut self, commit_shas: &[String]) -> Result<(), AutterError> { + if commit_shas.is_empty() { + return Ok(()); + } + let now = unix_now(); + let tx = self.conn.transaction()?; + for sha in commit_shas { + tx.execute( + r#"UPDATE notes + SET processing_started_at = NULL, + last_sync_error = ?1, + last_sync_at = ?2, + next_retry_at = ?2 + ?3 + WHERE commit_sha = ?4 AND synced = 0"#, + params![ + crate::api::client::REPO_NOT_CONNECTED, + now, + crate::api::client::UNCONNECTED_RECHECK_SECS, + sha + ], + )?; + } + tx.commit()?; + Ok(()) + } + + /// `(notes, commit_summaries)` kept on this machine because their repo + /// isn't connected to any of the user's orgs. + pub fn count_unconnected(&self) -> Result<(i64, i64), AutterError> { + let notes = self.conn.query_row( + "SELECT COUNT(*) FROM notes WHERE synced = 0 AND last_sync_error = ?1", + params![crate::api::client::REPO_NOT_CONNECTED], + |row| row.get(0), + )?; + let summaries = self.conn.query_row( + "SELECT COUNT(*) FROM commit_summary_queue WHERE last_sync_error = ?1", + params![crate::api::client::REPO_NOT_CONNECTED], + |row| row.get(0), + )?; + Ok((notes, summaries)) + } + // ----- Commit summary queue ----- pub fn enqueue_commit_summary( @@ -711,6 +758,29 @@ impl NotesDatabase { Ok(()) } + /// Park a commit summary whose repo isn't connected to any of the user's + /// orgs. See [`Self::defer_unconnected`]. + pub fn defer_unconnected_commit_summary( + &mut self, + commit_sha: &str, + ) -> Result<(), AutterError> { + let now = unix_now(); + self.conn.execute( + r#"UPDATE commit_summary_queue SET + processing_started_at = NULL, + last_sync_error = ?1, + next_retry_at = ?2 + ?3 + WHERE commit_sha = ?4"#, + params![ + crate::api::client::REPO_NOT_CONNECTED, + now, + crate::api::client::UNCONNECTED_RECHECK_SECS, + commit_sha + ], + )?; + Ok(()) + } + // ----- Read operations ----- /// Retrieve the note content for a single commit SHA. diff --git a/src/upload_hold.rs b/src/upload_hold.rs index b89c9bf..293a531 100644 --- a/src/upload_hold.rs +++ b/src/upload_hold.rs @@ -186,6 +186,30 @@ pub fn held_counts() -> QueueCounts { held } +/// Items kept on this machine because their repository isn't connected to any +/// of the user's orgs. Best-effort: an unavailable database counts as zero. +pub fn unconnected_counts() -> QueueCounts { + let mut kept = QueueCounts::default(); + if let Ok(db) = crate::notes::db::NotesDatabase::global() + && let Ok(db) = db.lock() + && let Ok((n, s)) = db.count_unconnected() + { + kept.notes = n; + kept.commit_summaries = s; + } + if let Ok(db) = crate::authorship::internal_db::InternalDatabase::global() + && let Ok(db) = db.lock() + { + kept.transcripts = db.count_unconnected_cas().unwrap_or(0); + } + if let Ok(db) = crate::file_changes::FileChangesDatabase::global() + && let Ok(db) = db.lock() + { + kept.file_changes = db.count_unconnected().unwrap_or(0); + } + kept +} + /// One-time migration: hold every item queued by autter 2.2.0 or earlier. /// /// Must run before anything drains a queue. Returns `Ok(true)` when the