diff --git a/.gitignore b/.gitignore index 6fdae7b..613d346 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,4 @@ PUBLISH.md tmp-statsai-*.sqlite* /statsai.sqlite* .wrangler/ +/.claude diff --git a/crates/statsai-core/src/lib.rs b/crates/statsai-core/src/lib.rs index 27789ff..3ff7cba 100644 --- a/crates/statsai-core/src/lib.rs +++ b/crates/statsai-core/src/lib.rs @@ -27,11 +27,13 @@ pub const DAILY_ROLLUP_SCHEMA_VERSION: &str = "daily_rollup.v1"; pub const SYNC_BATCH_V1_SCHEMA_VERSION: &str = "sync_batch.v1"; pub const SYNC_BATCH_V2_SCHEMA_VERSION: &str = "sync_batch.v2"; pub const SYNC_BATCH_V3_SCHEMA_VERSION: &str = "sync_batch.v3"; +pub const SYNC_BATCH_V4_SCHEMA_VERSION: &str = "sync_batch.v4"; pub const SYNC_ACK_V1_SCHEMA_VERSION: &str = "sync_ack.v1"; pub const SYNC_ACK_V2_SCHEMA_VERSION: &str = "sync_ack.v2"; pub const SYNC_ACK_V3_SCHEMA_VERSION: &str = "sync_ack.v3"; -pub const SYNC_BATCH_SCHEMA_VERSION: &str = SYNC_BATCH_V3_SCHEMA_VERSION; -pub const SYNC_ACK_SCHEMA_VERSION: &str = SYNC_ACK_V3_SCHEMA_VERSION; +pub const SYNC_ACK_V4_SCHEMA_VERSION: &str = "sync_ack.v4"; +pub const SYNC_BATCH_SCHEMA_VERSION: &str = SYNC_BATCH_V4_SCHEMA_VERSION; +pub const SYNC_ACK_SCHEMA_VERSION: &str = SYNC_ACK_V4_SCHEMA_VERSION; #[derive( Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema, @@ -727,6 +729,10 @@ pub struct SyncBatch { /// arguments, and commit messages are deliberately absent from this type. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub code_change_metrics: Vec, + /// Attributed quota-cycle contributions. Local quota records, payloads, + /// plans, credits, and sample counts are deliberately absent from this type. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub quota_cycle_contributions: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub authoritative_snapshot: Option, pub created_at: DateTime, @@ -749,6 +755,8 @@ pub struct SyncAuthoritativeSnapshot { pub summary_ids: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub code_change_metric_ids: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub quota_cycle_contribution_ids: Vec, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -796,6 +804,8 @@ pub struct SyncEntityCounts { pub task_verifications: u64, #[serde(default, skip_serializing_if = "sync_count_is_zero")] pub code_change_metrics: u64, + #[serde(default, skip_serializing_if = "sync_count_is_zero")] + pub quota_cycle_contributions: u64, } fn sync_count_is_zero(value: &u64) -> bool { @@ -2138,6 +2148,7 @@ mod tests { task_buckets: 0, task_verifications: 0, code_change_metrics: 0, + quota_cycle_contributions: 0, }, duplicates: SyncEntityCounts { sources: 0, @@ -2149,6 +2160,7 @@ mod tests { task_buckets: 0, task_verifications: 0, code_change_metrics: 0, + quota_cycle_contributions: 0, }, rejected: Vec::new(), }; @@ -2166,7 +2178,7 @@ mod tests { #[test] fn sync_ack_v3_keeps_nonzero_task_and_code_change_counters() { let ack = SyncAck { - schema_version: SYNC_ACK_SCHEMA_VERSION.to_string(), + schema_version: SYNC_ACK_V3_SCHEMA_VERSION.to_string(), batch_id: "batch-2".to_string(), accepted: SyncEntityCounts { sources: 0, @@ -2178,6 +2190,7 @@ mod tests { task_buckets: 3, task_verifications: 1, code_change_metrics: 2, + quota_cycle_contributions: 0, }, duplicates: SyncEntityCounts { sources: 0, @@ -2189,6 +2202,7 @@ mod tests { task_buckets: 0, task_verifications: 0, code_change_metrics: 0, + quota_cycle_contributions: 0, }, rejected: Vec::new(), }; @@ -2198,6 +2212,60 @@ mod tests { assert_eq!(json["accepted"]["task_buckets"], 3); assert_eq!(json["accepted"]["task_verifications"], 1); assert_eq!(json["accepted"]["code_change_metrics"], 2); + assert!(json["accepted"].get("quota_cycle_contributions").is_none()); + } + + #[test] + fn sync_ack_v4_keeps_nonzero_quota_cycle_counters() { + let ack = SyncAck { + schema_version: SYNC_ACK_V4_SCHEMA_VERSION.to_string(), + batch_id: "batch-quota".to_string(), + accepted: SyncEntityCounts { + sources: 0, + accounts: 0, + source_account_assignments: 0, + subscriptions: 0, + events: 0, + summaries: 0, + task_buckets: 0, + task_verifications: 0, + code_change_metrics: 1, + quota_cycle_contributions: 4, + }, + duplicates: SyncEntityCounts { + sources: 0, + accounts: 0, + source_account_assignments: 0, + subscriptions: 0, + events: 0, + summaries: 0, + task_buckets: 0, + task_verifications: 0, + code_change_metrics: 0, + quota_cycle_contributions: 0, + }, + rejected: Vec::new(), + }; + + let json = serde_json::to_value(&ack).expect("ack should serialize"); + assert_eq!(json["schema_version"], SYNC_ACK_V4_SCHEMA_VERSION); + assert_eq!(json["accepted"]["code_change_metrics"], 1); + assert_eq!(json["accepted"]["quota_cycle_contributions"], 4); + } + + #[test] + fn sync_batch_v3_without_quota_contributions_remains_backward_compatible() { + let batch: SyncBatch = serde_json::from_value(serde_json::json!({ + "schema_version": SYNC_BATCH_V3_SCHEMA_VERSION, + "batch_id": "batch-legacy-v3", + "device_id": "device-1", + "created_at": "2026-05-31T10:00:00Z" + })) + .expect("legacy v3 batch should deserialize"); + + assert!(batch.quota_cycle_contributions.is_empty()); + let serialized = serde_json::to_value(batch).expect("batch should serialize"); + assert!(serialized.get("quota_cycle_contributions").is_none()); } fn test_source(provider: &str, path: &str) -> SourceLocation { diff --git a/crates/statsai-core/src/quota.rs b/crates/statsai-core/src/quota.rs index 26c8238..ca590b9 100644 --- a/crates/statsai-core/src/quota.rs +++ b/crates/statsai-core/src/quota.rs @@ -10,6 +10,8 @@ pub const QUOTA_OBSERVATION_SCHEMA_VERSION: &str = "quota_observation.v1"; pub const QUOTA_WINDOW_OBSERVATION_SCHEMA_VERSION: &str = "quota_window_observation.v1"; pub const QUOTA_WINDOW_SCHEMA_VERSION: &str = "quota_window.v1"; pub const QUOTA_WINDOW_SYNC_PROJECTION_SCHEMA_VERSION: &str = "quota_window_sync_projection.v1"; +pub const QUOTA_CYCLE_CONTRIBUTION_SCHEMA_VERSION: &str = "quota_cycle_contribution.v1"; +pub const QUOTA_WEEKLY_WINDOW_MINUTES: u64 = 10_080; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "snake_case")] @@ -194,3 +196,134 @@ pub struct QuotaWindowSyncProjectionV1 { pub change_points: Vec, pub latest_status: QuotaProjectionStatusV1, } + +/// Timestamped first/last/min/max percentages observed on one UTC day. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +pub struct QuotaDailyEnvelopeV1 { + pub day: String, + pub first_observed_at: DateTime, + pub first_used_percent: f64, + pub last_observed_at: DateTime, + pub last_used_percent: f64, + pub minimum_used_percent: f64, + pub maximum_used_percent: f64, +} + +/// Exact usage for a partial UTC day at a cycle or schedule-transition boundary. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)] +pub struct QuotaUsageSliceV1 { + pub period_start: DateTime, + pub period_end: DateTime, + pub input_tokens: u64, + pub cache_creation_tokens: u64, + pub cache_read_tokens: u64, + pub output_tokens: u64, + pub reasoning_tokens: u64, + pub total_tokens: u64, + pub estimated_cost_micro_usd: i64, +} + +impl QuotaUsageSliceV1 { + pub fn add_usage(&mut self, usage: &UsageCounts, estimated_cost_micro_usd: Option) { + self.input_tokens = self + .input_tokens + .saturating_add(usage.input_tokens.unwrap_or(0)); + // The 5m and 1h counters are the lifetime breakdown of the aggregate, + // not additional writes: Claude reports 248 alongside 148 + 100. Adding + // all three counted every cached write twice and pushed the slice + // components past its own `total_tokens`. Every other accumulator in + // the workspace takes the aggregate alone; the lifetimes are only a + // fallback for records that arrive without it. + self.cache_creation_tokens = + self.cache_creation_tokens + .saturating_add(usage.cache_creation_tokens.unwrap_or_else(|| { + usage + .cache_creation_5m_tokens + .unwrap_or(0) + .saturating_add(usage.cache_creation_1h_tokens.unwrap_or(0)) + })); + self.cache_read_tokens = self + .cache_read_tokens + .saturating_add(usage.cache_read_tokens.unwrap_or(0)); + self.output_tokens = self + .output_tokens + .saturating_add(usage.output_tokens.unwrap_or(0)); + self.reasoning_tokens = self + .reasoning_tokens + .saturating_add(usage.reasoning_tokens.unwrap_or(0)); + self.total_tokens = self.total_tokens.saturating_add(usage.computed_total()); + if let Some(value) = estimated_cost_micro_usd { + self.estimated_cost_micro_usd = self.estimated_cost_micro_usd.saturating_add(value); + } + } +} + +/// Hosted device contribution for one attributed quota cycle. Omits raw events, +/// paths, source IDs, payloads, plans, credits, slots, fingerprints, and sample +/// counts. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +pub struct QuotaCycleContributionV1 { + pub schema_version: String, + /// Deterministic device contribution ID, not a logical account-cycle ID. + pub contribution_id: String, + pub provider: String, + pub provider_account_id: ProviderAccountId, + pub limit_id: Option, + pub window_minutes: u64, + pub representative_reset: DateTime, + pub representative_reset_epoch_seconds: i64, + /// True when this device locally reconstructed another cycle for the same + /// scope whose schedule overlaps this one. Codex weekly cycles start lazily + /// at first use, so a corroborated overlap means the neighbouring cycle was + /// reset early (banked or server-granted), not that the data conflicts. + #[serde(default)] + pub has_schedule_overlap: bool, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub daily_envelopes: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub boundary_slices: Vec, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_slice_counts_each_cached_write_once() { + // Claude reports the aggregate beside its lifetime breakdown, and the + // two describe the same 248 tokens. + let usage = UsageCounts { + input_tokens: Some(10), + output_tokens: Some(20), + cache_creation_tokens: Some(248), + cache_creation_5m_tokens: Some(148), + cache_creation_1h_tokens: Some(100), + total_tokens: Some(278), + ..UsageCounts::default() + }; + + let mut slice = QuotaUsageSliceV1::default(); + slice.add_usage(&usage, None); + + assert_eq!(slice.cache_creation_tokens, 248); + assert!( + slice.input_tokens + slice.output_tokens + slice.cache_creation_tokens + <= slice.total_tokens + ); + } + + #[test] + fn a_slice_falls_back_to_the_lifetime_breakdown() { + let usage = UsageCounts { + cache_creation_tokens: None, + cache_creation_5m_tokens: Some(8), + cache_creation_1h_tokens: Some(5), + ..UsageCounts::default() + }; + + let mut slice = QuotaUsageSliceV1::default(); + slice.add_usage(&usage, None); + + assert_eq!(slice.cache_creation_tokens, 13); + } +} diff --git a/crates/statsai-daemon/src/lib.rs b/crates/statsai-daemon/src/lib.rs index b63e95e..6e30c7d 100644 --- a/crates/statsai-daemon/src/lib.rs +++ b/crates/statsai-daemon/src/lib.rs @@ -4,8 +4,9 @@ use anyhow::{bail, Context, Result}; use serde_json::json; use statsai_core::{ SyncAck, SyncBatch, SyncEntityCounts, SyncRejectedRecord, SYNC_ACK_V1_SCHEMA_VERSION, - SYNC_ACK_V2_SCHEMA_VERSION, SYNC_ACK_V3_SCHEMA_VERSION, SYNC_BATCH_V1_SCHEMA_VERSION, - SYNC_BATCH_V2_SCHEMA_VERSION, SYNC_BATCH_V3_SCHEMA_VERSION, + SYNC_ACK_V2_SCHEMA_VERSION, SYNC_ACK_V3_SCHEMA_VERSION, SYNC_ACK_V4_SCHEMA_VERSION, + SYNC_BATCH_V1_SCHEMA_VERSION, SYNC_BATCH_V2_SCHEMA_VERSION, SYNC_BATCH_V3_SCHEMA_VERSION, + SYNC_BATCH_V4_SCHEMA_VERSION, }; use statsai_store::Store; use std::io::Read; @@ -30,6 +31,7 @@ fn sync_ack_schema_version(batch_schema_version: &str) -> Result<&'static str> { SYNC_BATCH_V1_SCHEMA_VERSION => Ok(SYNC_ACK_V1_SCHEMA_VERSION), SYNC_BATCH_V2_SCHEMA_VERSION => Ok(SYNC_ACK_V2_SCHEMA_VERSION), SYNC_BATCH_V3_SCHEMA_VERSION => Ok(SYNC_ACK_V3_SCHEMA_VERSION), + SYNC_BATCH_V4_SCHEMA_VERSION => Ok(SYNC_ACK_V4_SCHEMA_VERSION), other => bail!("unsupported sync batch schema {other}"), } } @@ -219,13 +221,22 @@ pub fn ingest_sync_batch(store: &Store, batch: &SyncBatch) -> Result { if batch.schema_version != SYNC_BATCH_V1_SCHEMA_VERSION && batch.schema_version != SYNC_BATCH_V2_SCHEMA_VERSION && batch.schema_version != SYNC_BATCH_V3_SCHEMA_VERSION + && batch.schema_version != SYNC_BATCH_V4_SCHEMA_VERSION { bail!("unsupported sync batch schema {}", batch.schema_version); } - if batch.schema_version != SYNC_BATCH_V3_SCHEMA_VERSION && !batch.code_change_metrics.is_empty() + if !matches!( + batch.schema_version.as_str(), + SYNC_BATCH_V3_SCHEMA_VERSION | SYNC_BATCH_V4_SCHEMA_VERSION + ) && !batch.code_change_metrics.is_empty() { bail!("code-change metrics require sync_batch.v3"); } + if batch.schema_version != SYNC_BATCH_V4_SCHEMA_VERSION + && !batch.quota_cycle_contributions.is_empty() + { + bail!("quota cycle contributions require sync_batch.v4"); + } if batch .code_change_metrics .iter() @@ -236,6 +247,16 @@ pub fn ingest_sync_batch(store: &Store, batch: &SyncBatch) -> Result { if batch.authoritative_snapshot.is_some() { bail!("authoritative snapshots are not supported by the loopback daemon"); } + // A local store holds quota observations and derives its own cycles from + // them; there is no table for another device's contributions. Acknowledging + // them as accepted told the sender to record them synced against this + // target and stop offering them, certifying a write that never happened. + // Refusing follows the authoritative-snapshot precedent above: this + // endpoint is for loopback diagnostics, and `/api/sync/batches` is the + // contract that stores quota cycles. + if !batch.quota_cycle_contributions.is_empty() { + bail!("quota cycle contributions are not supported by the loopback daemon"); + } let result = store.ingest_sync_batch(batch)?; @@ -252,6 +273,7 @@ pub fn ingest_sync_batch(store: &Store, batch: &SyncBatch) -> Result { task_buckets: batch.task_buckets.len() as u64, task_verifications: result.merged_task_verifications, code_change_metrics: batch.code_change_metrics.len() as u64, + quota_cycle_contributions: batch.quota_cycle_contributions.len() as u64, }, duplicates: SyncEntityCounts { sources: 0, @@ -264,6 +286,7 @@ pub fn ingest_sync_batch(store: &Store, batch: &SyncBatch) -> Result { task_verifications: (batch.task_verifications.len() as u64) .saturating_sub(result.merged_task_verifications), code_change_metrics: 0, + quota_cycle_contributions: 0, }, rejected: Vec::::new(), }) @@ -2682,6 +2705,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: Utc::now(), } @@ -2841,6 +2865,62 @@ mod tests { .contains("code-change metrics require sync_batch.v3")); } + #[test] + fn ingest_v3_batch_rejects_quota_cycle_contributions() { + let store = Store::in_memory().expect("store"); + let mut batch = empty_batch(); + batch.schema_version = SYNC_BATCH_V3_SCHEMA_VERSION.to_string(); + batch + .quota_cycle_contributions + .push(statsai_core::QuotaCycleContributionV1 { + schema_version: statsai_core::QUOTA_CYCLE_CONTRIBUTION_SCHEMA_VERSION.to_string(), + contribution_id: "quota_cycle_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(), + provider: "codex".to_string(), + provider_account_id: statsai_core::ProviderAccountId("acct-1".to_string()), + limit_id: None, + window_minutes: statsai_core::QUOTA_WEEKLY_WINDOW_MINUTES, + representative_reset: Utc::now(), + representative_reset_epoch_seconds: Utc::now().timestamp(), + has_schedule_overlap: false, + daily_envelopes: Vec::new(), + boundary_slices: Vec::new(), + }); + + let error = ingest_sync_batch(&store, &batch).expect_err("v3 quota payload"); + assert!(error + .to_string() + .contains("quota cycle contributions require sync_batch.v4")); + } + + #[test] + fn ingest_v4_batch_refuses_quota_cycle_contributions_it_cannot_store() { + let store = Store::in_memory().expect("store"); + let mut batch = empty_batch(); + batch.schema_version = SYNC_BATCH_V4_SCHEMA_VERSION.to_string(); + batch + .quota_cycle_contributions + .push(statsai_core::QuotaCycleContributionV1 { + schema_version: statsai_core::QUOTA_CYCLE_CONTRIBUTION_SCHEMA_VERSION.to_string(), + contribution_id: "quota_cycle_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(), + provider: "codex".to_string(), + provider_account_id: statsai_core::ProviderAccountId("acct-1".to_string()), + limit_id: None, + window_minutes: statsai_core::QUOTA_WEEKLY_WINDOW_MINUTES, + representative_reset: Utc::now(), + representative_reset_epoch_seconds: Utc::now().timestamp(), + has_schedule_overlap: false, + daily_envelopes: Vec::new(), + boundary_slices: Vec::new(), + }); + + // Accepting them would tell the sender to stop offering cycles this + // store has nowhere to put. + let error = ingest_sync_batch(&store, &batch).expect_err("unsupported contributions"); + assert!(error + .to_string() + .contains("quota cycle contributions are not supported")); + } + #[test] fn ingest_v3_batch_rejects_metric_owned_by_another_device_before_persisting() { let store = Store::in_memory().expect("store"); diff --git a/crates/statsai-dev/src/github.rs b/crates/statsai-dev/src/github.rs index 91be89f..04622e6 100644 --- a/crates/statsai-dev/src/github.rs +++ b/crates/statsai-dev/src/github.rs @@ -12,6 +12,20 @@ pub(crate) const REPOSITORY: &str = "starkdmi/statsai"; pub(crate) const WORKFLOW_FILE: &str = "dev-build.yml"; const GITHUB_API: &str = "https://api.github.com"; const MAX_ARTIFACT_BYTES: u64 = 256 * 1024 * 1024; +/// Enough for a JSON reply, and short enough that an unreachable API gives up +/// promptly. +const API_TIMEOUT: Duration = Duration::from_secs(30); +/// `ureq`'s request timeout covers reading the body too, so the API bound also +/// capped every artifact download. Thirty seconds for an archive that may reach +/// the 256 MiB ceiling demands roughly 68 Mbit/s sustained, which turned an +/// ordinary home connection into an intermittent "timed out reading response". +/// Ten minutes covers the ceiling at about 3.5 Mbit/s. +const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(600); +/// Overrides [`DOWNLOAD_TIMEOUT`], in seconds. Zero waits indefinitely. +const DOWNLOAD_TIMEOUT_ENV: &str = "STATSAI_DEV_DOWNLOAD_TIMEOUT_SECONDS"; +/// One initial attempt plus two retries. +const MAX_DOWNLOAD_ATTEMPTS: u32 = 3; +const DOWNLOAD_RETRY_BASE_DELAY: Duration = Duration::from_secs(2); #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum BuildRequest { @@ -250,7 +264,7 @@ impl GitHubClient { let mut downloads = Vec::with_capacity(artifacts.len()); let mut errors = Vec::new(); for artifact in artifacts { - match self.download(&artifact.archive_download_url, &expected_name) { + match self.download_with_retries(&artifact.archive_download_url, &expected_name) { Ok(bytes) => downloads.push(bytes), Err(error) => errors.push(format!("artifact #{}: {error:#}", artifact.id)), } @@ -265,16 +279,34 @@ impl GitHubClient { Ok(downloads) } + /// The wait loop that precedes this retries the *build lookup*, not the + /// download, so a transfer cut short by a slow link failed the whole install + /// on the first try. A body cannot be resumed, so a retry re-requests it. + fn download_with_retries(&self, url: &str, expected_name: &str) -> Result> { + retrying_download( + || self.download(url, expected_name), + std::thread::sleep, + |message| eprintln!("{message}"), + ) + } + fn download(&self, url: &str, expected_name: &str) -> Result> { let response = self.call( - self.request(url), + self.request_with_timeout(url, download_timeout()), &format!("download artifact {expected_name}"), )?; let mut reader = response.into_reader().take(MAX_ARTIFACT_BYTES + 1); let mut bytes = Vec::new(); - reader - .read_to_end(&mut bytes) - .with_context(|| format!("read downloaded artifact {expected_name}"))?; + // A read that dies partway is as transient as one that never connected, + // but it surfaces here rather than from `call`, so it has to be marked + // retryable explicitly or the loop above would give up on it. + if let Err(error) = reader.read_to_end(&mut bytes) { + return Err(GitHubRequestFailure { + message: format!("read downloaded artifact {expected_name}: {error}"), + retry: Some(RetryAdvice { delay: None }), + } + .into()); + } if bytes.len() as u64 > MAX_ARTIFACT_BYTES { bail!("artifact {expected_name} exceeds the 256 MiB safety limit"); } @@ -282,8 +314,16 @@ impl GitHubClient { } fn request(&self, url: &str) -> ureq::Request { - let request = ureq::get(url) - .timeout(Duration::from_secs(30)) + self.request_with_timeout(url, Some(API_TIMEOUT)) + } + + fn request_with_timeout(&self, url: &str, timeout: Option) -> ureq::Request { + let request = ureq::get(url); + let request = match timeout { + Some(timeout) => request.timeout(timeout), + None => request, + }; + let request = request .set( "User-Agent", &format!("statsai-dev/{}", env!("CARGO_PKG_VERSION")), @@ -407,6 +447,46 @@ fn compare_runs(left: &WorkflowRun, right: &WorkflowRun) -> Ordering { .then_with(|| left.id.cmp(&right.id)) } +fn retrying_download(mut download: D, mut sleep: S, mut warn: W) -> Result> +where + D: FnMut() -> Result>, + S: FnMut(Duration), + W: FnMut(String), +{ + let mut attempt = 1; + loop { + let error = match download() { + Ok(bytes) => return Ok(bytes), + Err(error) => error, + }; + let Some(retry) = retry_advice(&error) else { + return Err(error); + }; + if attempt >= MAX_DOWNLOAD_ATTEMPTS { + return Err(error.context(format!("download failed {MAX_DOWNLOAD_ATTEMPTS} times"))); + } + let delay = retry.delay.unwrap_or(DOWNLOAD_RETRY_BASE_DELAY * attempt); + warn(format!( + "warning: {error:#}; retrying download ({attempt} of {}) in {} seconds", + MAX_DOWNLOAD_ATTEMPTS - 1, + delay.as_secs() + )); + sleep(delay); + attempt += 1; + } +} + +fn download_timeout() -> Option { + let Ok(raw) = std::env::var(DOWNLOAD_TIMEOUT_ENV) else { + return Some(DOWNLOAD_TIMEOUT); + }; + match raw.trim().parse::() { + Ok(0) => None, + Ok(seconds) => Some(Duration::from_secs(seconds)), + Err(_) => Some(DOWNLOAD_TIMEOUT), + } +} + fn github_token() -> Option { for variable in ["STATSAI_DEV_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"] { if let Ok(value) = std::env::var(variable) { @@ -499,6 +579,108 @@ const fn default_attempt() -> u64 { #[cfg(test)] mod tests { + use std::sync::{Mutex, OnceLock}; + + fn env_lock() -> &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + } + + #[test] + fn a_cut_off_download_is_retried_rather_than_failing_the_install() { + // The wait loop retries the build lookup, not the transfer, so before + // this a single interrupted read ended the whole install. + let mut attempts = 0; + let mut slept = Vec::new(); + let mut warnings = Vec::new(); + let bytes = retrying_download( + || { + attempts += 1; + if attempts < 3 { + Err(GitHubRequestFailure { + message: "timed out reading response".to_string(), + retry: Some(RetryAdvice { delay: None }), + } + .into()) + } else { + Ok(vec![7, 7, 7]) + } + }, + |delay| slept.push(delay), + |message| warnings.push(message), + ) + .expect("third attempt succeeds"); + + assert_eq!(bytes, vec![7, 7, 7]); + assert_eq!(attempts, 3); + assert_eq!( + slept, + vec![DOWNLOAD_RETRY_BASE_DELAY, DOWNLOAD_RETRY_BASE_DELAY * 2] + ); + assert_eq!(warnings.len(), 2); + assert!(warnings[0].contains("retrying download")); + } + + #[test] + fn a_download_gives_up_after_the_attempt_limit() { + let mut attempts = 0; + let error = retrying_download( + || { + attempts += 1; + Err(GitHubRequestFailure { + message: "timed out reading response".to_string(), + retry: Some(RetryAdvice { delay: None }), + } + .into()) + }, + |_| {}, + |_| {}, + ) + .expect_err("always fails"); + + assert_eq!(attempts, MAX_DOWNLOAD_ATTEMPTS); + assert!(format!("{error:#}").contains("download failed")); + } + + #[test] + fn an_oversized_artifact_is_not_retried() { + // Re-requesting cannot make it smaller. + let mut attempts = 0; + let error = retrying_download( + || { + attempts += 1; + bail!("artifact exceeds the 256 MiB safety limit") + }, + |_| panic!("must not sleep"), + |_| panic!("must not warn"), + ) + .expect_err("permanent failure"); + + assert_eq!(attempts, 1); + assert!(format!("{error:#}").contains("safety limit")); + } + + #[test] + fn the_download_timeout_is_generous_and_overridable() { + let _guard = env_lock().lock().unwrap_or_else(|error| error.into_inner()); + std::env::remove_var(DOWNLOAD_TIMEOUT_ENV); + // A 256 MiB artifact cannot arrive inside the API timeout on an + // ordinary connection, which is what made downloads fail at random. + assert_eq!(download_timeout(), Some(DOWNLOAD_TIMEOUT)); + assert!(DOWNLOAD_TIMEOUT > API_TIMEOUT); + + std::env::set_var(DOWNLOAD_TIMEOUT_ENV, "45"); + assert_eq!(download_timeout(), Some(Duration::from_secs(45))); + + std::env::set_var(DOWNLOAD_TIMEOUT_ENV, "0"); + assert_eq!(download_timeout(), None, "zero waits indefinitely"); + + std::env::set_var(DOWNLOAD_TIMEOUT_ENV, "not-a-number"); + assert_eq!(download_timeout(), Some(DOWNLOAD_TIMEOUT)); + + std::env::remove_var(DOWNLOAD_TIMEOUT_ENV); + } + use super::*; fn run( diff --git a/crates/statsai-store/src/code_changes.rs b/crates/statsai-store/src/code_changes.rs index a2e9308..5c70b4e 100644 --- a/crates/statsai-store/src/code_changes.rs +++ b/crates/statsai-store/src/code_changes.rs @@ -2166,6 +2166,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: vec![remote_metric.clone()], + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: Utc::now(), }) diff --git a/crates/statsai-store/src/lib.rs b/crates/statsai-store/src/lib.rs index e6ff279..2bb135d 100644 --- a/crates/statsai-store/src/lib.rs +++ b/crates/statsai-store/src/lib.rs @@ -107,6 +107,7 @@ pub struct PendingSyncSummaryCounts { pub rollups: u64, pub passthrough_summaries: u64, pub retired_entities: u64, + pub quota_cycle_contributions: u64, pub total: u64, pub days: u64, } @@ -2510,6 +2511,11 @@ impl Store { target, &batch.code_change_metrics, )?; + self.record_quota_cycle_contributions_synced_in_transaction( + sink, + target, + &batch.quota_cycle_contributions, + )?; self.record_task_bucket_snapshots_synced_in_transaction( sink, target, @@ -2936,6 +2942,14 @@ impl Store { .map(String::as_str) .collect::>(), ), + ( + "quota_cycle_contribution", + snapshot + .quota_cycle_contribution_ids + .iter() + .map(String::as_str) + .collect::>(), + ), ]); let mut statement = self.conn.prepare( r#" @@ -2944,7 +2958,7 @@ impl Store { WHERE sink = ?1 AND target = ?2 AND entity_kind IN ( 'source', 'account', 'source_account_assignment', 'subscription', 'summary', - 'code_change_metric' + 'code_change_metric', 'quota_cycle_contribution' ) "#, )?; @@ -3019,10 +3033,25 @@ impl Store { target, ¤t_code_change_metrics, )?; + let current_quota_cycle_contributions = + self.quota_cycle_contributions(&QuotaQuery::default(), device_id)?; let current_snapshot = self.current_http_sync_authoritative_snapshot( ¤t_rollups, ¤t_passthrough_summaries, ¤t_code_change_metrics, + ¤t_quota_cycle_contributions + .iter() + .map(|contribution| contribution.contribution_id.clone()) + .collect::>(), + )?; + // A quota cycle can change without any summary changing: a reset moves, + // or an observation carries no tokens. Counting only the summary-shaped + // entities left those uploads invisible, so the menubar reported nothing + // pending while a sync would still have sent them. + let quota_cycle_contributions = self.pending_quota_cycle_contributions_for_sync( + "http", + target, + ¤t_quota_cycle_contributions, )?; let retired_entities = self .retired_sync_entity_ids("http", target, ¤t_snapshot)? @@ -3034,10 +3063,12 @@ impl Store { rollups: rollups.len() as u64, passthrough_summaries: passthrough_summaries.len() as u64, retired_entities: retired_entities as u64, + quota_cycle_contributions: quota_cycle_contributions.len() as u64, total: rollups .len() .saturating_add(passthrough_summaries.len()) .saturating_add(code_change_metrics.len()) + .saturating_add(quota_cycle_contributions.len()) .saturating_add(retired_entities) as u64, days: days.len() as u64, }) @@ -3048,6 +3079,7 @@ impl Store { rollups: &[UsageSummary], passthrough_summaries: &[UsageSummary], code_change_metrics: &[CodeChangeMetric], + quota_cycle_contribution_ids: &[String], ) -> Result { Ok(SyncAuthoritativeSnapshot { snapshot_id: String::new(), @@ -3082,6 +3114,7 @@ impl Store { .iter() .map(|metric| metric.metric_id.clone()) .collect(), + quota_cycle_contribution_ids: quota_cycle_contribution_ids.to_vec(), }) } @@ -3497,6 +3530,39 @@ impl Store { Ok(()) } + pub fn record_quota_cycle_contributions_synced( + &self, + sink: &str, + target: &str, + contributions: &[statsai_core::QuotaCycleContributionV1], + ) -> Result<()> { + if contributions.is_empty() { + return Ok(()); + } + self.with_immediate_transaction(|| { + self.record_quota_cycle_contributions_synced_in_transaction(sink, target, contributions) + }) + } + + fn record_quota_cycle_contributions_synced_in_transaction( + &self, + sink: &str, + target: &str, + contributions: &[statsai_core::QuotaCycleContributionV1], + ) -> Result<()> { + for contribution in contributions { + let payload = serde_json::to_string(contribution)?; + self.record_entity_synced( + sink, + target, + "quota_cycle_contribution", + &contribution.contribution_id, + &hash_text(&payload), + )?; + } + Ok(()) + } + pub fn compute_daily_rollup(&self, date: &str, device_id: &str) -> Result { let mut stmt = self.conn.prepare( r#" @@ -8619,6 +8685,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: now, }; @@ -9795,6 +9862,7 @@ mod tests { rollups: 0, passthrough_summaries: 2, retired_entities: 0, + quota_cycle_contributions: 0, total: 2, days: 5, } @@ -9856,6 +9924,7 @@ mod tests { rollups: 0, passthrough_summaries: 1, retired_entities: 0, + quota_cycle_contributions: 0, total: 1, days: 1, } diff --git a/crates/statsai-store/src/quota.rs b/crates/statsai-store/src/quota.rs index b88a5d0..a615822 100644 --- a/crates/statsai-store/src/quota.rs +++ b/crates/statsai-store/src/quota.rs @@ -1,19 +1,33 @@ use super::{assignment_for_timestamp, Store}; use anyhow::Result; -use chrono::{DateTime, Duration, Utc}; +use chrono::{DateTime, Duration, Timelike, Utc}; use rusqlite::{params, OptionalExtension}; use serde::Serialize; use statsai_core::{ hash_text, periods_overlap, EventId, ProviderAccountId, QuotaChangePointV1, - QuotaObservationRecordV1, QuotaObservationV1, QuotaProjectionStatusV1, QuotaTransitionKind, - QuotaUsageLinkKind, QuotaUsageTotalsV1, QuotaWindowObservationV1, QuotaWindowSyncProjectionV1, - QuotaWindowV1, SourceAccountAssignment, SourceId, QUOTA_WINDOW_SCHEMA_VERSION, + QuotaCycleContributionV1, QuotaDailyEnvelopeV1, QuotaObservationRecordV1, QuotaObservationV1, + QuotaProjectionStatusV1, QuotaTransitionKind, QuotaUsageLinkKind, QuotaUsageSliceV1, + QuotaUsageTotalsV1, QuotaWindowObservationV1, QuotaWindowSyncProjectionV1, QuotaWindowV1, + SourceAccountAssignment, SourceId, QUOTA_CYCLE_CONTRIBUTION_SCHEMA_VERSION, + QUOTA_WEEKLY_WINDOW_MINUTES, QUOTA_WINDOW_SCHEMA_VERSION, QUOTA_WINDOW_SYNC_PROJECTION_SCHEMA_VERSION, }; use std::collections::{BTreeMap, HashMap, HashSet}; const RESET_CLUSTER_TOLERANCE_SECONDS: i64 = 5 * 60; -pub const SYNC_ELIGIBLE_WINDOW_MINUTES: u64 = 10_080; +/// A window cannot be observed once it has reset: the provider issues a fresh +/// `resets_at` from that moment on. An observation that post-dates its own reset +/// by more than provider recomputation lag is therefore a replay of historical +/// evidence recorded at import time, and must not extend the closed window. +/// +/// The bound is an hour because the provider has been seen serving an elapsed +/// window for 37 continuous minutes: roughly 180 separate requests, seconds +/// apart, every one still carrying the expired `resets_at` at a steady 96%. +/// Discarding those loses the closing figure of a cycle spent to exhaustion. +/// Genuine lag has never been observed to outlast that; the stale records past +/// it sit 14 hours to 3 days late, which is import time, not lag. +const STALE_OBSERVATION_TOLERANCE_SECONDS: i64 = 60 * 60; +pub const SYNC_ELIGIBLE_WINDOW_MINUTES: u64 = QUOTA_WEEKLY_WINDOW_MINUTES; #[derive(Debug, Clone, Default)] pub struct QuotaQuery { @@ -44,9 +58,27 @@ pub struct QuotaStatus { pub weekly_observations: u64, pub weekly_sync_eligible_observations: u64, pub weekly_sync_eligible_coverage_percent: f64, + pub discarded: QuotaDiscardCounts, pub assignment_overlap_warnings: Vec, } +/// What reconstruction threw away, and why. +/// +/// Each rule discards evidence that cannot describe a real cycle, and each is +/// silent by design. Counting the discards is what tells a provider change +/// apart from a quiet week: if these numbers climb, the rules have started +/// firing on data they were never meant to judge. +#[derive(Debug, Clone, Default, PartialEq, Serialize)] +pub struct QuotaDiscardCounts { + /// Observations recorded more than an hour after the window they describe + /// had already reset. + pub replayed_observations: u64, + /// Windows that reported zero for their whole life and were superseded. + pub unused_windows: u64, + /// Schedules another schedule was reported both before and after. + pub bracketed_schedules: u64, +} + impl Store { pub fn clear_orphaned_quota_usage_links(&self) -> Result { Ok(self.conn.execute( @@ -421,6 +453,7 @@ impl Store { } pub fn quota_status(&self, query: &QuotaQuery) -> Result { + let (_, discarded) = self.reconstruct_quota_windows_counted(query)?; let raw = self.quota_observations(query, false)?; let total_observations = raw.len(); let distinct = collapse_semantic_duplicates(raw); @@ -438,7 +471,7 @@ impl Store { record .windows .iter() - .any(|window| window.window_minutes >= SYNC_ELIGIBLE_WINDOW_MINUTES) + .any(|window| window.window_minutes == SYNC_ELIGIBLE_WINDOW_MINUTES) }) .collect::>(); let weekly_eligible = weekly @@ -484,6 +517,7 @@ impl Store { } else { weekly_eligible as f64 * 100.0 / weekly.len() as f64 }, + discarded, assignment_overlap_warnings: warnings, }) } @@ -498,6 +532,26 @@ impl Store { &self, query: &QuotaQuery, ) -> Result> { + Ok(self + .reconstruct_quota_windows(query)? + .into_iter() + .map(|reconstructed| reconstructed.window) + .collect()) + } + + fn reconstruct_quota_windows( + &self, + query: &QuotaQuery, + ) -> Result> { + let (windows, _) = self.reconstruct_quota_windows_counted(query)?; + Ok(windows) + } + + fn reconstruct_quota_windows_counted( + &self, + query: &QuotaQuery, + ) -> Result<(Vec, QuotaDiscardCounts)> { + let mut discarded = QuotaDiscardCounts::default(); let mut reconstruction_query = query.clone(); reconstruction_query.from = None; reconstruction_query.to = None; @@ -505,6 +559,10 @@ impl Store { let mut grouped = BTreeMap::>::new(); for record in records { for window in &record.windows { + if observation_postdates_reset(&record.observation, window) { + discarded.replayed_observations += 1; + continue; + } grouped .entry(WindowScope { provider: record.observation.provider.clone(), @@ -533,14 +591,22 @@ impl Store { point.observation.observed_at, ) }); + // The tolerance is a gap between neighbouring resets, not a width + // the whole cycle must fit inside. A cycle's reported reset drifts + // a second or two at a time and has been seen wandering nine + // minutes over its life; measured against the cluster's first + // point that cycle is cut in two the moment it passes five, and + // the remainder is drawn as a second cycle running beside the + // first over the same days. Consecutive resets sit hours apart at + // the very least, so chaining cannot reach across one. let mut clusters: Vec> = Vec::new(); for point in points { if let Some(cluster) = clusters.last_mut() { - let min_reset = cluster - .first() + let previous_reset = cluster + .last() .map(|point| point.window.resets_at_epoch_seconds) .unwrap_or(point.window.resets_at_epoch_seconds); - if point.window.resets_at_epoch_seconds - min_reset + if point.window.resets_at_epoch_seconds - previous_reset <= RESET_CLUSTER_TOLERANCE_SECONDS { cluster.push(point); @@ -550,6 +616,34 @@ impl Store { clusters.push(vec![point]); } + // A window that reported zero usage for its whole life and was then + // replaced never functioned as a cycle: nothing was ever attributed + // to it, so it cannot be told apart from a reschedule of the window + // that followed. Provider-side resets during the July 2026 tier + // migration left runs of these minutes apart, each one drawn as a + // separate reset that never happened. The final cluster is exempt, + // because a cycle that has only just begun legitimately reads zero. + let cluster_count = clusters.len(); + let clusters = clusters + .into_iter() + .enumerate() + .filter(|(index, cluster)| { + *index + 1 == cluster_count + || cluster.iter().any(|point| point.window.used_percent > 0.0) + }) + .map(|(_, cluster)| cluster) + .collect::>(); + discarded.unused_windows += (cluster_count - clusters.len()) as u64; + + let phantoms = phantom_cluster_indices(&clusters); + discarded.bracketed_schedules += phantoms.len() as u64; + let clusters = clusters + .into_iter() + .enumerate() + .filter(|(index, _)| !phantoms.contains(index)) + .map(|(_, cluster)| cluster) + .collect::>(); + let schedules = clusters .iter() .map(|cluster| QuotaClusterSchedule::from_points(&scope, cluster)) @@ -561,6 +655,7 @@ impl Store { if !quota_cluster_matches_time_query(schedule, query) { continue; } + let daily_envelopes = daily_envelopes_from_points(&cluster); let mut window = Self::materialize_quota_window(&scope, schedule, cluster); window.transition = if index == 0 { QuotaTransitionKind::Initial @@ -574,18 +669,22 @@ impl Store { && schedule.inferred_start < peer.representative_reset && schedule.representative_reset > peer.inferred_start }); - scope_windows.push(window); + scope_windows.push(ReconstructedQuotaWindow { + window, + daily_envelopes, + }); } windows.extend(scope_windows); } windows.sort_by(|left, right| { right + .window .representative_reset - .cmp(&left.representative_reset) - .then_with(|| right.window_minutes.cmp(&left.window_minutes)) - .then_with(|| left.window_id.cmp(&right.window_id)) + .cmp(&left.window.representative_reset) + .then_with(|| right.window.window_minutes.cmp(&left.window.window_minutes)) + .then_with(|| left.window.window_id.cmp(&right.window.window_id)) }); - Ok(windows) + Ok((windows, discarded)) } fn materialize_quota_window( @@ -762,7 +861,7 @@ impl Store { Ok(self .quota_windows_without_usage_totals(query)? .into_iter() - .filter(|window| window.window_minutes >= SYNC_ELIGIBLE_WINDOW_MINUTES) + .filter(|window| window.window_minutes == SYNC_ELIGIBLE_WINDOW_MINUTES) .filter_map(|window| { let provider_account_id = window.provider_account_id.clone()?; let anchor = window.change_points.first()?; @@ -805,6 +904,167 @@ impl Store { }) .collect()) } + + pub fn quota_cycle_contributions( + &self, + query: &QuotaQuery, + device_id: &str, + ) -> Result> { + let reconstructed = self.reconstruct_quota_windows(query)?; + let mut contributions = Vec::new(); + let mut pending = Vec::new(); + for item in reconstructed { + if item.window.window_minutes != QUOTA_WEEKLY_WINDOW_MINUTES { + continue; + } + let Some(provider_account_id) = item.window.provider_account_id.clone() else { + continue; + }; + let Some(anchor) = item.window.change_points.first() else { + continue; + }; + let contribution_id = format!( + "quota_cycle_{}", + &hash_text(&format!( + "quota_cycle_contribution.v1:{device_id}:{}:{}:{}:{}:{}", + item.window.provider, + provider_account_id.0, + item.window.limit_id.as_deref().unwrap_or("default"), + item.window.window_minutes, + anchor.point_fingerprint + ))[..32] + ); + pending.push(PendingQuotaCycleContribution { + contribution_id, + provider_account_id, + reconstructed: item, + }); + } + + let effective_bounds = effective_cycle_bounds( + pending + .iter() + .map(|item| &item.reconstructed.window) + .collect::>(), + ); + let mut slices_by_window = HashMap::>::new(); + if !pending.is_empty() { + let first_start = pending + .iter() + .map(|item| { + effective_bounds + .get(&item.reconstructed.window.window_id) + .map(|bounds| bounds.start) + .unwrap_or(item.reconstructed.window.inferred_start) + }) + .min() + .expect("pending contributions are non-empty"); + let last_end = pending + .iter() + .map(|item| { + effective_bounds + .get(&item.reconstructed.window.window_id) + .map(|bounds| bounds.end) + .unwrap_or(item.reconstructed.window.representative_reset) + }) + .max() + .expect("pending contributions are non-empty"); + let mut slice_builders = pending + .iter() + .map(|item| { + let bounds = effective_bounds + .get(&item.reconstructed.window.window_id) + .copied() + .unwrap_or(CycleBounds { + start: item.reconstructed.window.inferred_start, + end: item.reconstructed.window.representative_reset, + }); + ( + item.reconstructed.window.window_id.clone(), + boundary_slice_builders( + bounds.start, + bounds.end, + &item.reconstructed.window.provider, + item.reconstructed + .window + .provider_account_id + .as_ref() + .expect("attributed cycle"), + ), + ) + }) + .collect::>(); + + for event in self.events_in_period(Some(first_start), last_end)? { + let Some(account_id) = event.provider_account_id.as_ref() else { + continue; + }; + let estimated_cost = event.cost.estimated_micro_usd(); + for (_, builders) in &mut slice_builders { + if event.provider != builders.provider || account_id != &builders.account_id { + continue; + } + for slice in builders.slices.iter_mut() { + if event.session.started_at < slice.period_start + || event.session.started_at >= slice.period_end + { + continue; + } + slice.add_usage(&event.usage, estimated_cost); + } + } + } + + for (window_id, builders) in slice_builders { + slices_by_window.insert(window_id, builders.slices); + } + } + + for item in pending { + let boundary_slices = slices_by_window + .remove(&item.reconstructed.window.window_id) + .unwrap_or_default(); + contributions.push(QuotaCycleContributionV1 { + schema_version: QUOTA_CYCLE_CONTRIBUTION_SCHEMA_VERSION.to_string(), + contribution_id: item.contribution_id, + provider: item.reconstructed.window.provider, + provider_account_id: item.provider_account_id, + limit_id: item.reconstructed.window.limit_id, + window_minutes: item.reconstructed.window.window_minutes, + representative_reset: item.reconstructed.window.representative_reset, + representative_reset_epoch_seconds: item + .reconstructed + .window + .representative_reset_epoch_seconds, + has_schedule_overlap: item.reconstructed.window.has_schedule_overlap, + daily_envelopes: item.reconstructed.daily_envelopes, + boundary_slices, + }); + } + Ok(contributions) + } + + pub fn pending_quota_cycle_contributions_for_sync( + &self, + sink: &str, + target: &str, + contributions: &[QuotaCycleContributionV1], + ) -> Result> { + let mut pending = Vec::new(); + for contribution in contributions { + let payload = serde_json::to_string(contribution)?; + if self.entity_requires_sync( + sink, + target, + "quota_cycle_contribution", + &contribution.contribution_id, + &hash_text(&payload), + )? { + pending.push(contribution.clone()); + } + } + Ok(pending) + } } #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] @@ -822,6 +1082,69 @@ struct WindowPoint { window: QuotaWindowObservationV1, } +/// The span over which a cluster was observed. +struct ClusterLife { + first_observed_at: DateTime, + last_observed_at: DateTime, +} + +fn cluster_life(cluster: &[WindowPoint]) -> Option { + Some(ClusterLife { + first_observed_at: cluster.iter().map(|p| p.observation.observed_at).min()?, + last_observed_at: cluster.iter().map(|p| p.observation.observed_at).max()?, + }) +} + +/// True when `live` was reported before `suspect` appeared and again after it +/// vanished. +fn brackets(live: &[WindowPoint], suspect: &ClusterLife) -> bool { + live.iter() + .any(|p| p.observation.observed_at < suspect.first_observed_at) + && live + .iter() + .any(|p| p.observation.observed_at > suspect.last_observed_at) +} + +/// Clusters that describe a schedule the provider never actually switched to. +/// +/// A reset is a handover: once a window resets, the provider reports the new +/// schedule and never returns to the old one. So a cluster that another cluster +/// brackets in observation time — reported before this one appeared and again +/// after it vanished — cannot be a cycle in its own right, however many +/// observations back it. +/// +/// During the July 2026 tier migration Codex answered a scattering of turns +/// with a blank snapshot: near-zero usage and a fresh weekly reset, interleaved +/// with the live schedule for as long as 34 hours. Each run became a cycle that +/// took days away from the cycle actually running, which is what put two rising +/// lines over the same hours for a single account. +/// +/// Bracketing alone settles it, whatever the two schedules claim was spent. +/// Once a window resets, its `resets_at` moves forward and that value can never +/// be reported again, so a schedule cannot resume after another has taken over +/// from it. A genuine early reset — banked or granted — is therefore never +/// bracketed: the schedule it replaced simply stops. +/// +/// This deliberately drops well-evidenced clusters. One such ran to 99% over +/// 19 hours across 623 observations while the schedule either side of it sat +/// near 54%, which is not a cycle that reset early but a second counter +/// reported under the same slot for the same account. +fn phantom_cluster_indices(clusters: &[Vec]) -> HashSet { + let lives = clusters.iter().map(|c| cluster_life(c)).collect::>(); + let mut phantoms = HashSet::new(); + for (index, life) in lives.iter().enumerate() { + let Some(life) = life else { continue }; + let bracketed = clusters + .iter() + .enumerate() + .any(|(other, live)| other != index && brackets(live, life)); + if bracketed { + phantoms.insert(index); + } + } + phantoms +} + #[derive(Debug, Clone)] struct QuotaClusterSchedule { inferred_start: DateTime, @@ -864,6 +1187,212 @@ impl QuotaClusterSchedule { } } +#[derive(Debug, Clone)] +struct ReconstructedQuotaWindow { + window: QuotaWindowV1, + daily_envelopes: Vec, +} + +struct PendingQuotaCycleContribution { + contribution_id: String, + provider_account_id: ProviderAccountId, + reconstructed: ReconstructedQuotaWindow, +} + +#[derive(Debug, Clone, Copy)] +struct CycleBounds { + start: DateTime, + end: DateTime, +} + +struct BoundarySliceBuilders { + provider: String, + account_id: ProviderAccountId, + slices: Vec, +} + +/// True when the evidence was recorded after the window it describes had already +/// reset, which only happens when historical records are re-imported and stamped +/// with the import time. Such a point would otherwise be clustered into the long +/// closed cycle by its `resets_at` and then bucketed into a daily envelope by its +/// `observed_at`, giving that cycle an observation days or weeks past its end. +fn observation_postdates_reset( + observation: &QuotaObservationV1, + window: &QuotaWindowObservationV1, +) -> bool { + (observation.observed_at - window.resets_at).num_seconds() > STALE_OBSERVATION_TOLERANCE_SECONDS +} + +fn daily_envelopes_from_points(points: &[WindowPoint]) -> Vec { + let mut grouped = BTreeMap::>::new(); + for point in points { + grouped + .entry(point.observation.observed_at.date_naive().to_string()) + .or_default() + .push(point); + } + let mut envelopes = grouped + .into_iter() + .map(|(day, mut day_points)| { + day_points.sort_by_key(|point| point.observation.observed_at); + let first = day_points.first().expect("non-empty utc day"); + let last = day_points.last().expect("non-empty utc day"); + QuotaDailyEnvelopeV1 { + day, + first_observed_at: first.observation.observed_at, + first_used_percent: first.window.used_percent, + last_observed_at: last.observation.observed_at, + last_used_percent: last.window.used_percent, + minimum_used_percent: day_points + .iter() + .map(|point| point.window.used_percent) + .fold(f64::INFINITY, f64::min), + maximum_used_percent: day_points + .iter() + .map(|point| point.window.used_percent) + .fold(f64::NEG_INFINITY, f64::max), + } + }) + .collect::>(); + + // Consumption cannot fall inside a window, so a reading below one already + // seen is stale rather than a refund. Concurrent sessions produce them + // routinely: two poll at once and the slower one still holds the earlier + // figure, sometimes within the same second. Left alone the daily closing + // walks backwards. The observed extremes stay untouched, since they are + // the evidence that the jitter happened. + let mut consumed = f64::NEG_INFINITY; + for envelope in &mut envelopes { + if consumed.is_finite() { + envelope.first_used_percent = envelope.first_used_percent.max(consumed); + } + consumed = consumed.max(envelope.maximum_used_percent); + envelope.last_used_percent = envelope.last_used_percent.max(consumed); + } + envelopes +} + +fn utc_day_start(timestamp: DateTime) -> DateTime { + timestamp + .date_naive() + .and_hms_opt(0, 0, 0) + .expect("midnight") + .and_utc() +} + +fn is_utc_midnight(timestamp: DateTime) -> bool { + timestamp.time().num_seconds_from_midnight() == 0 && timestamp.nanosecond() == 0 +} + +fn next_utc_day_start(timestamp: DateTime) -> DateTime { + utc_day_start(timestamp) + Duration::days(1) +} + +fn empty_usage_slice(period_start: DateTime, period_end: DateTime) -> QuotaUsageSliceV1 { + QuotaUsageSliceV1 { + period_start, + period_end, + ..QuotaUsageSliceV1::default() + } +} + +fn boundary_slice_builders( + start: DateTime, + end: DateTime, + provider: &str, + account_id: &ProviderAccountId, +) -> BoundarySliceBuilders { + let mut slices = Vec::new(); + if start >= end { + return BoundarySliceBuilders { + provider: provider.to_string(), + account_id: account_id.clone(), + slices, + }; + } + if start.date_naive() == end.date_naive() { + if !is_utc_midnight(start) || !is_utc_midnight(end) { + slices.push(empty_usage_slice(start, end)); + } + return BoundarySliceBuilders { + provider: provider.to_string(), + account_id: account_id.clone(), + slices, + }; + } + if !is_utc_midnight(start) { + slices.push(empty_usage_slice(start, next_utc_day_start(start))); + } + if !is_utc_midnight(end) { + slices.push(empty_usage_slice(utc_day_start(end), end)); + } + BoundarySliceBuilders { + provider: provider.to_string(), + account_id: account_id.clone(), + slices, + } +} + +fn clamp_datetime(value: DateTime, min: DateTime, max: DateTime) -> DateTime { + value.max(min).min(max) +} + +fn effective_cycle_bounds(windows: Vec<&QuotaWindowV1>) -> HashMap { + let mut grouped = BTreeMap::<(String, String, Option, u64), Vec<&QuotaWindowV1>>::new(); + for window in windows { + let Some(account_id) = window.provider_account_id.as_ref() else { + continue; + }; + grouped + .entry(( + window.provider.clone(), + account_id.0.clone(), + window.limit_id.clone(), + window.window_minutes, + )) + .or_default() + .push(window); + } + let mut bounds = HashMap::new(); + for mut scope_windows in grouped.into_values() { + scope_windows.sort_by_key(|window| { + ( + window.inferred_start, + window.representative_reset, + window.window_id.clone(), + ) + }); + for window in &scope_windows { + bounds.insert( + window.window_id.clone(), + CycleBounds { + start: window.inferred_start, + end: window.representative_reset, + }, + ); + } + for index in 1..scope_windows.len() { + let previous = scope_windows[index - 1]; + let current = scope_windows[index]; + let overlap_start = previous.inferred_start.max(current.inferred_start); + let overlap_end = previous + .representative_reset + .min(current.representative_reset); + if overlap_start >= overlap_end { + continue; + } + let transition = clamp_datetime(current.first_observed_at, overlap_start, overlap_end); + if let Some(previous_bounds) = bounds.get_mut(&previous.window_id) { + previous_bounds.end = previous_bounds.end.min(transition); + } + if let Some(current_bounds) = bounds.get_mut(¤t.window_id) { + current_bounds.start = current_bounds.start.max(transition); + } + } + } + bounds +} + fn quota_point_fingerprint(scope: &WindowScope, point: &WindowPoint) -> String { let identity_scope = scope.provider_account_id.as_ref().map_or_else( || { @@ -1056,9 +1585,12 @@ fn parse_usage_link_kind(value: &str) -> QuotaUsageLinkKind { #[cfg(test)] mod tests { use super::*; + use serde_json::json; use statsai_core::{ - IdentitySource, LocationOrigin, QuotaCreditsV1, QuotaStatusV1, SourceAccountAssignment, - SourceAccountAssignmentId, SourceLocation, SOURCE_ACCOUNT_ASSIGNMENT_SCHEMA_VERSION, + event_id, Confidence, CostInfo, EventSource, IdentitySource, LocationOrigin, PrivacyInfo, + PrivacyMode, QuotaCreditsV1, QuotaStatusV1, SessionInfo, SourceAccountAssignment, + SourceAccountAssignmentId, SourceKind, SourceLocation, UsageCounts, UsageEvent, + SOURCE_ACCOUNT_ASSIGNMENT_SCHEMA_VERSION, USAGE_EVENT_SCHEMA_VERSION, }; use std::collections::HashSet; use std::path::Path; @@ -1902,4 +2434,970 @@ mod tests { assert!(projection_json.get("total_tokens").is_none()); assert!(projection_json.get("estimated_cost").is_none()); } + + #[allow(clippy::too_many_arguments)] + fn sample_usage_event( + source_id: &SourceId, + account_id: &ProviderAccountId, + started_at: DateTime, + record_id: &str, + input_tokens: u64, + cache_read_tokens: u64, + output_tokens: u64, + reasoning_tokens: u64, + estimated_cost_micro_usd: i64, + ) -> UsageEvent { + UsageEvent { + schema_version: USAGE_EVENT_SCHEMA_VERSION.to_string(), + event_id: event_id("codex", source_id, record_id, None, started_at), + device_id: "device".to_string(), + provider: "codex".to_string(), + source_id: source_id.clone(), + provider_account_id: Some(account_id.clone()), + subscription_id: None, + source: EventSource { + adapter_id: "test".to_string(), + adapter_version: "0".to_string(), + source_kind: SourceKind::LocalAdapter, + location_origin: Some(LocationOrigin::Configured), + source_type: "jsonl".to_string(), + source_path_hash: Some("quota-event".to_string()), + source_record_id: Some(record_id.to_string()), + parse_confidence: Confidence::High, + }, + session: SessionInfo { + session_id: record_id.to_string(), + local_session_id_hash: Some(record_id.to_string()), + title: None, + started_at, + ended_at: None, + duration_seconds: None, + }, + model: None, + usage: UsageCounts { + input_tokens: Some(input_tokens), + output_tokens: Some(output_tokens), + cache_read_tokens: Some(cache_read_tokens), + reasoning_tokens: Some(reasoning_tokens), + total_tokens: Some( + input_tokens + cache_read_tokens + output_tokens + reasoning_tokens, + ), + ..UsageCounts::default() + }, + runtime: None, + cost: CostInfo { + currency: "USD".to_string(), + estimated_api_equivalent_usd: None, + provider_reported_usd: None, + estimated_api_equivalent_micro_usd: Some(estimated_cost_micro_usd), + provider_reported_micro_usd: None, + pricing_source: Some("test".to_string()), + pricing_version: None, + confidence: Confidence::High, + }, + parse_evidence: None, + project: None, + git: None, + privacy: PrivacyInfo { + mode: PrivacyMode::MetadataOnly, + contains_prompt_text: false, + contains_response_text: false, + contains_file_paths: false, + }, + created_at: started_at, + imported_at: started_at, + } + } + + #[test] + fn a_quota_only_change_still_reports_pending_upload_work() { + // Nothing here writes a summary: the pending count has to see the + // contribution itself or the menubar claims there is nothing to send. + let store = Store::in_memory().expect("store"); + let observed_at = DateTime::from_timestamp(1_787_011_200, 0).expect("observed"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let (source_id, _) = assigned_source(&store, observed_at - Duration::days(8)); + store + .upsert_quota_observations(&[sample_record( + source_id, + "weekly", + "weekly", + observed_at, + reset.timestamp(), + "secondary", + 10_080, + 20.0, + )]) + .expect("observations"); + + let target = "https://api.example.com/api/sync/batches"; + let counts = store + .pending_http_sync_summary_counts(target, "device-a") + .expect("counts"); + assert_eq!(counts.rollups, 0); + assert_eq!(counts.passthrough_summaries, 0); + assert_eq!(counts.quota_cycle_contributions, 1); + assert!( + counts.total > 0, + "a quota-only change is pending upload work" + ); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + store + .record_quota_cycle_contributions_synced("http", target, &contributions) + .expect("record synced"); + + let settled = store + .pending_http_sync_summary_counts(target, "device-a") + .expect("settled counts"); + assert_eq!(settled.quota_cycle_contributions, 0); + } + + #[test] + fn quota_cycle_contributions_select_weekly_attributed_cycles_only() { + let store = Store::in_memory().expect("store"); + let observed_at = DateTime::from_timestamp(1_787_011_200, 0).expect("observed"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let (source_id, _) = assigned_source(&store, observed_at - Duration::days(8)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "weekly", + "weekly", + observed_at, + reset.timestamp(), + "secondary", + 10_080, + 20.0, + ), + sample_record( + source_id.clone(), + "five-hour", + "five-hour", + observed_at, + (reset - Duration::hours(4)).timestamp(), + "primary", + 300, + 40.0, + ), + sample_record( + source_id, + "monthly", + "monthly", + observed_at, + (reset + Duration::days(20)).timestamp(), + "monthly", + 43_200, + 8.0, + ), + ]) + .expect("observations"); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + assert_eq!(contributions.len(), 1); + assert_eq!(contributions[0].window_minutes, 10_080); + assert_eq!( + contributions[0].schema_version, + "quota_cycle_contribution.v1" + ); + let json = serde_json::to_value(&contributions[0]).expect("json"); + assert!(json.get("source_id").is_none()); + assert!(json.get("device_id").is_none()); + assert!(json.get("change_points").is_none()); + assert!(json.get("sample_count").is_none()); + assert!(json.get("latest_status").is_none()); + assert_eq!(json.get("has_schedule_overlap"), Some(&json!(false))); + } + + #[test] + fn stale_concurrent_readings_never_walk_the_daily_closing_backwards() { + let store = Store::in_memory().expect("store"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let day_one = reset - Duration::days(3); + let day_two = day_one + Duration::days(1); + let (source_id, _) = assigned_source(&store, reset - Duration::days(9)); + // Two sessions poll together; the slower one still reports the older + // figure, and on the second day it lands last. + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "d1-low", + "d1-low", + day_one, + reset.timestamp(), + "secondary", + 10_080, + 27.0, + ), + sample_record( + source_id.clone(), + "d1-high", + "d1-high", + day_one + Duration::hours(2), + reset.timestamp(), + "secondary", + 10_080, + 59.0, + ), + sample_record( + source_id, + "d2-stale", + "d2-stale", + day_two, + reset.timestamp(), + "secondary", + 10_080, + 56.0, + ), + ]) + .expect("observations"); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + assert_eq!(contributions.len(), 1); + let envelopes = &contributions[0].daily_envelopes; + assert_eq!(envelopes.len(), 2); + // The stale 56% must not read as a drop from the 59% already spent. + assert_eq!(envelopes[0].last_used_percent, 59.0); + assert_eq!(envelopes[1].last_used_percent, 59.0); + assert_eq!(envelopes[1].first_used_percent, 59.0); + // The raw extremes still record that a 56% reading arrived. + assert_eq!(envelopes[1].minimum_used_percent, 56.0); + assert!( + envelopes + .windows(2) + .all(|pair| pair[1].last_used_percent >= pair[0].last_used_percent), + "daily closings never decrease inside a cycle" + ); + } + + #[test] + fn a_window_that_never_left_zero_is_not_a_cycle() { + let store = Store::in_memory().expect("store"); + let real = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + // The provider briefly reported a window resetting 40 minutes earlier, + // at zero, before settling on the one that went on to be used. + let stub_reset = real - Duration::minutes(40); + let (source_id, _) = assigned_source(&store, real - Duration::days(9)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "stub", + "stub", + stub_reset - Duration::days(7) + Duration::minutes(5), + stub_reset.timestamp(), + "secondary", + 10_080, + 0.0, + ), + sample_record( + source_id, + "real", + "real", + real - Duration::days(6), + real.timestamp(), + "secondary", + 10_080, + 64.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!( + windows.len(), + 1, + "the zero-only window is not its own cycle" + ); + assert_eq!(windows[0].representative_reset, real); + } + + #[test] + fn status_reports_what_reconstruction_threw_away() { + let store = Store::in_memory().expect("store"); + let live_reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let start = live_reset - Duration::days(7); + let (source_id, _) = assigned_source(&store, start - Duration::days(2)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "live-1", + "live-1", + start + Duration::hours(1), + live_reset.timestamp(), + "secondary", + 10_080, + 21.0, + ), + // Bracketed by the live schedule, so not a cycle of its own. + sample_record( + source_id.clone(), + "phantom", + "phantom", + start + Duration::hours(2), + (live_reset + Duration::days(1)).timestamp(), + "secondary", + 10_080, + 1.0, + ), + sample_record( + source_id.clone(), + "live-2", + "live-2", + start + Duration::hours(3), + live_reset.timestamp(), + "secondary", + 10_080, + 44.0, + ), + // Recorded a day after the window it describes had reset. + sample_record( + source_id, + "replayed", + "replayed", + live_reset + Duration::days(1), + live_reset.timestamp(), + "secondary", + 10_080, + 44.0, + ), + ]) + .expect("observations"); + + let status = store.quota_status(&QuotaQuery::default()).expect("status"); + assert_eq!(status.discarded.replayed_observations, 1); + assert_eq!(status.discarded.bracketed_schedules, 1); + } + + #[test] + fn an_interleaved_schedule_the_provider_never_switched_to_is_not_a_cycle() { + let store = Store::in_memory().expect("store"); + let live_reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let start = live_reset - Duration::days(7); + // Codex answered two turns mid-cycle with a blank snapshot: a token + // percentage against a reset a day later than the one it kept + // reporting either side. + let phantom_reset = live_reset + Duration::days(1); + let (source_id, _) = assigned_source(&store, start - Duration::days(2)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "live-1", + "live-1", + start + Duration::hours(1), + live_reset.timestamp(), + "secondary", + 10_080, + 21.0, + ), + sample_record( + source_id.clone(), + "phantom-1", + "phantom-1", + start + Duration::hours(2), + phantom_reset.timestamp(), + "secondary", + 10_080, + 0.0, + ), + sample_record( + source_id.clone(), + "phantom-2", + "phantom-2", + start + Duration::hours(3), + phantom_reset.timestamp(), + "secondary", + 10_080, + 1.0, + ), + sample_record( + source_id, + "live-2", + "live-2", + start + Duration::hours(4), + live_reset.timestamp(), + "secondary", + 10_080, + 44.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!( + windows.len(), + 1, + "a schedule reported between two readings of the one that outran it did not reset" + ); + assert_eq!(windows[0].representative_reset, live_reset); + assert_eq!(windows[0].maximum_used_percent, 44.0); + } + + #[test] + fn an_early_reset_that_took_over_stays_a_cycle() { + let store = Store::in_memory().expect("store"); + let first_reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let start = first_reset - Duration::days(7); + // A redeemed reset zeroes the window and issues a fresh schedule. The + // schedule it replaced is never reported again, so nothing brackets it. + let redeemed_reset = first_reset + Duration::days(4); + let (source_id, _) = assigned_source(&store, start - Duration::days(2)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "before-1", + "before-1", + start + Duration::hours(1), + first_reset.timestamp(), + "secondary", + 10_080, + 21.0, + ), + sample_record( + source_id.clone(), + "before-2", + "before-2", + start + Duration::hours(2), + first_reset.timestamp(), + "secondary", + 10_080, + 64.0, + ), + sample_record( + source_id.clone(), + "after-1", + "after-1", + start + Duration::hours(3), + redeemed_reset.timestamp(), + "secondary", + 10_080, + 0.0, + ), + sample_record( + source_id, + "after-2", + "after-2", + start + Duration::hours(4), + redeemed_reset.timestamp(), + "secondary", + 10_080, + 9.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!( + windows.len(), + 2, + "an early reset that the provider switched to is its own cycle" + ); + assert!(windows + .iter() + .any(|window| window.representative_reset == redeemed_reset)); + } + + #[test] + fn a_cycle_that_has_only_just_begun_survives_at_zero() { + let store = Store::in_memory().expect("store"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let (source_id, _) = assigned_source(&store, reset - Duration::days(9)); + store + .upsert_quota_observations(&[sample_record( + source_id, + "fresh", + "fresh", + reset - Duration::days(7) + Duration::minutes(5), + reset.timestamp(), + "secondary", + 10_080, + 0.0, + )]) + .expect("observation"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!(windows.len(), 1, "the newest cycle is exempt"); + assert_eq!(windows[0].maximum_used_percent, 0.0); + } + + #[test] + fn a_superseded_window_that_spent_anything_stays_a_cycle() { + let store = Store::in_memory().expect("store"); + let real = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let earlier_reset = real - Duration::minutes(40); + let (source_id, _) = assigned_source(&store, real - Duration::days(9)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "spent-a-little", + "spent-a-little", + earlier_reset - Duration::days(7) + Duration::minutes(5), + earlier_reset.timestamp(), + "secondary", + 10_080, + 3.0, + ), + sample_record( + source_id, + "real", + "real", + real - Duration::days(6), + real.timestamp(), + "secondary", + 10_080, + 64.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!(windows.len(), 2, "3% spent is still a real cycle"); + } + + #[test] + fn replayed_observations_do_not_extend_a_window_that_already_reset() { + let store = Store::in_memory().expect("store"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let live = reset - Duration::days(2); + // A re-import three weeks later re-reads the same historical window and + // stamps it with the import time. + let replay = reset + Duration::days(21); + let (source_id, _) = assigned_source(&store, live - Duration::days(8)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "live", + "live", + live, + reset.timestamp(), + "secondary", + 10_080, + 26.0, + ), + sample_record( + source_id, + "replayed", + "replayed", + replay, + reset.timestamp(), + "secondary", + 10_080, + 58.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!(windows.len(), 1); + let days = windows[0] + .change_points + .iter() + .map(|point| point.observed_at.date_naive().to_string()) + .collect::>(); + assert_eq!( + days, + HashSet::from([live.date_naive().to_string()]), + "the closed window keeps only evidence recorded while it was open" + ); + assert_eq!(windows[0].maximum_used_percent, 26.0); + } + + #[test] + fn an_observation_within_recomputation_lag_still_belongs_to_its_window() { + let store = Store::in_memory().expect("store"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let (source_id, _) = assigned_source(&store, reset - Duration::days(8)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "before", + "before", + reset - Duration::hours(1), + reset.timestamp(), + "secondary", + 10_080, + 90.0, + ), + // The provider has not recomputed the window yet, so it keeps + // reporting the reset that elapsed 45 minutes ago. + sample_record( + source_id, + "lagging", + "lagging", + reset + Duration::minutes(45), + reset.timestamp(), + "secondary", + 10_080, + 100.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!(windows.len(), 1); + assert_eq!(windows[0].maximum_used_percent, 100.0); + assert_eq!(windows[0].sample_count, 2); + } + + #[test] + fn an_observation_beyond_recomputation_lag_is_treated_as_a_replay() { + let store = Store::in_memory().expect("store"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let (source_id, _) = assigned_source(&store, reset - Duration::days(8)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "live", + "live", + reset - Duration::hours(2), + reset.timestamp(), + "secondary", + 10_080, + 90.0, + ), + sample_record( + source_id, + "past-lag", + "past-lag", + reset + Duration::minutes(61), + reset.timestamp(), + "secondary", + 10_080, + 100.0, + ), + ]) + .expect("observations"); + + let windows = store + .quota_windows(&QuotaQuery::default()) + .expect("windows"); + assert_eq!(windows.len(), 1); + assert_eq!(windows[0].sample_count, 1); + assert_eq!(windows[0].maximum_used_percent, 90.0); + } + + #[test] + fn quota_cycle_contributions_report_locally_reconstructed_schedule_overlaps() { + let store = Store::in_memory().expect("store"); + // An early reset restarts the weekly schedule three days in, so the two + // reconstructed cycles overlap for the remaining four days. + let first_reset = DateTime::from_timestamp(1_787_616_000, 0).expect("first reset"); + let second_reset = first_reset + Duration::days(3); + let first_start = first_reset - Duration::days(7); + let (source_id, _) = assigned_source(&store, first_start - Duration::days(1)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "weekly-first", + "weekly-first", + first_start + Duration::hours(1), + first_reset.timestamp(), + "secondary", + 10_080, + 62.0, + ), + sample_record( + source_id, + "weekly-second", + "weekly-second", + second_reset - Duration::days(7) + Duration::hours(1), + second_reset.timestamp(), + "secondary", + 10_080, + 4.0, + ), + ]) + .expect("observations"); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + assert_eq!(contributions.len(), 2); + assert!( + contributions + .iter() + .all(|contribution| contribution.has_schedule_overlap), + "both sides of a locally reconstructed overlap carry the flag" + ); + } + + #[test] + fn quota_cycle_contributions_default_schedule_overlap_when_absent_on_the_wire() { + let wire = json!({ + "schema_version": "quota_cycle_contribution.v1", + "contribution_id": "quota_cycle_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "provider": "codex", + "provider_account_id": "acct_aaaaaaaaaaaaaaaaaaaaaaaa", + "limit_id": "codex", + "window_minutes": 10_080, + "representative_reset": "2026-08-25T15:00:00Z", + "representative_reset_epoch_seconds": 1_787_670_000i64, + }); + let contribution: QuotaCycleContributionV1 = + serde_json::from_value(wire).expect("legacy payload deserializes"); + assert!(!contribution.has_schedule_overlap); + } + + #[test] + fn quota_cycle_contributions_exclude_unattributed_cycles() { + let store = Store::in_memory().expect("store"); + let observed_at = DateTime::from_timestamp(1_787_011_200, 0).expect("observed"); + let source = SourceLocation::local_adapter( + "codex", + "codex-local-jsonl", + "unattributed", + Path::new("/tmp/quota-unattributed-cycle"), + LocationOrigin::Configured, + ); + store.upsert_source(&source).expect("source"); + store + .upsert_quota_observations(&[sample_record( + source.source_id, + "unattributed-weekly", + "unattributed-weekly", + observed_at, + 1_787_616_000, + "secondary", + 10_080, + 15.0, + )]) + .expect("observation"); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + assert!(contributions.is_empty()); + } + + #[test] + fn quota_cycle_contributions_build_daily_envelopes_without_carry_forward() { + let store = Store::in_memory().expect("store"); + let day_one = DateTime::from_timestamp(1_787_011_200, 0).expect("day one"); + let day_three = day_one + Duration::days(2) + Duration::hours(3); + let reset = day_one + Duration::days(7); + let (source_id, _) = assigned_source(&store, day_one - Duration::days(1)); + store + .upsert_quota_observations(&[ + sample_record( + source_id.clone(), + "day-one-first", + "day-one-first", + day_one + Duration::hours(1), + reset.timestamp(), + "secondary", + 10_080, + 10.0, + ), + sample_record( + source_id.clone(), + "day-one-last", + "day-one-last", + day_one + Duration::hours(8), + reset.timestamp(), + "secondary", + 10_080, + 25.0, + ), + sample_record( + source_id, + "day-three", + "day-three", + day_three, + reset.timestamp(), + "secondary", + 10_080, + 40.0, + ), + ]) + .expect("observations"); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + assert_eq!(contributions.len(), 1); + let days = contributions[0] + .daily_envelopes + .iter() + .map(|envelope| envelope.day.as_str()) + .collect::>(); + assert_eq!(days, ["2026-08-18", "2026-08-20"]); + assert_eq!(contributions[0].daily_envelopes[0].first_used_percent, 10.0); + assert_eq!(contributions[0].daily_envelopes[0].last_used_percent, 25.0); + assert_eq!( + contributions[0].daily_envelopes[0].minimum_used_percent, + 10.0 + ); + assert_eq!( + contributions[0].daily_envelopes[0].maximum_used_percent, + 25.0 + ); + assert_eq!(contributions[0].daily_envelopes[1].first_used_percent, 40.0); + assert_eq!(contributions[0].daily_envelopes[1].last_used_percent, 40.0); + } + + #[test] + fn quota_cycle_contributions_emit_exact_boundary_slices() { + let store = Store::in_memory().expect("store"); + // Use a mid-day reset so start and end fall on partial UTC days. + let reset = DateTime::from_timestamp(1_787_670_000, 0).expect("reset"); + let inferred_start = reset - Duration::minutes(10_080); + let observed_at = inferred_start + Duration::hours(2); + let (source_id, account_id) = assigned_source(&store, inferred_start - Duration::days(1)); + store + .upsert_quota_observations(&[sample_record( + source_id.clone(), + "weekly-boundary", + "weekly-boundary", + observed_at, + reset.timestamp(), + "secondary", + 10_080, + 33.0, + )]) + .expect("observation"); + store + .insert_event(&sample_usage_event( + &source_id, + &account_id, + inferred_start + Duration::minutes(30), + "start-boundary", + 100, + 20, + 10, + 5, + 1_500, + )) + .expect("start event"); + store + .insert_event(&sample_usage_event( + &source_id, + &account_id, + utc_day_start(reset) + Duration::hours(1), + "end-boundary", + 40, + 0, + 8, + 2, + 700, + )) + .expect("end event"); + store + .insert_event(&sample_usage_event( + &source_id, + &account_id, + inferred_start + Duration::days(1) + Duration::hours(2), + "interior-day", + 9_999, + 0, + 0, + 0, + 99_000, + )) + .expect("interior event"); + + let contributions = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("contributions"); + assert_eq!(contributions.len(), 1); + assert_eq!(contributions[0].boundary_slices.len(), 2); + assert_eq!( + contributions[0].boundary_slices[0].period_start, + inferred_start + ); + assert_eq!( + contributions[0].boundary_slices[0].period_end, + next_utc_day_start(inferred_start) + ); + assert_eq!(contributions[0].boundary_slices[0].input_tokens, 100); + assert_eq!(contributions[0].boundary_slices[0].cache_read_tokens, 20); + assert_eq!(contributions[0].boundary_slices[0].output_tokens, 10); + assert_eq!(contributions[0].boundary_slices[0].reasoning_tokens, 5); + assert_eq!(contributions[0].boundary_slices[0].total_tokens, 135); + assert_eq!( + contributions[0].boundary_slices[0].estimated_cost_micro_usd, + 1_500 + ); + assert_eq!( + contributions[0].boundary_slices[1].period_start, + utc_day_start(reset) + ); + assert_eq!(contributions[0].boundary_slices[1].period_end, reset); + assert_eq!(contributions[0].boundary_slices[1].input_tokens, 40); + assert_eq!( + contributions[0].boundary_slices[1].estimated_cost_micro_usd, + 700 + ); + assert!( + contributions[0] + .boundary_slices + .iter() + .all(|slice| slice.input_tokens != 9_999), + "complete utc days stay out of boundary slices" + ); + } + + #[test] + fn quota_cycle_contribution_ids_are_stable_for_the_same_device_anchor() { + let store = Store::in_memory().expect("store"); + let observed_at = DateTime::from_timestamp(1_787_011_200, 0).expect("observed"); + let reset = DateTime::from_timestamp(1_787_616_000, 0).expect("reset"); + let (source_id, _) = assigned_source(&store, observed_at - Duration::days(8)); + store + .upsert_quota_observations(&[sample_record( + source_id, + "stable-id", + "stable-id", + observed_at, + reset.timestamp(), + "secondary", + 10_080, + 18.0, + )]) + .expect("observation"); + + let first = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("first"); + let second = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-a") + .expect("second"); + let peer = store + .quota_cycle_contributions(&QuotaQuery::default(), "device-b") + .expect("peer"); + assert_eq!(first.len(), 1); + assert_eq!(first[0].contribution_id, second[0].contribution_id); + assert_ne!(first[0].contribution_id, peer[0].contribution_id); + assert!(first[0].contribution_id.starts_with("quota_cycle_")); + assert_eq!(first[0].contribution_id.len(), "quota_cycle_".len() + 32); + } } diff --git a/crates/statsai-sync/src/lib.rs b/crates/statsai-sync/src/lib.rs index adcdc2f..ad981f1 100644 --- a/crates/statsai-sync/src/lib.rs +++ b/crates/statsai-sync/src/lib.rs @@ -3,8 +3,8 @@ use anyhow::{bail, Context, Result}; use statsai_core::{ SyncAck, SyncBatch, SYNC_ACK_V1_SCHEMA_VERSION, SYNC_ACK_V2_SCHEMA_VERSION, - SYNC_ACK_V3_SCHEMA_VERSION, SYNC_BATCH_V1_SCHEMA_VERSION, SYNC_BATCH_V2_SCHEMA_VERSION, - SYNC_BATCH_V3_SCHEMA_VERSION, + SYNC_ACK_V3_SCHEMA_VERSION, SYNC_ACK_V4_SCHEMA_VERSION, SYNC_BATCH_V1_SCHEMA_VERSION, + SYNC_BATCH_V2_SCHEMA_VERSION, SYNC_BATCH_V3_SCHEMA_VERSION, SYNC_BATCH_V4_SCHEMA_VERSION, }; use std::io::Write; use std::path::{Path, PathBuf}; @@ -91,6 +91,9 @@ impl HttpSink { Ok(response) => response, Err(ureq::Error::Status(code, response)) => { let body = response.into_string().unwrap_or_default(); + if let Some(detail) = unrecognized_field_detail(&body) { + bail!("sync endpoint returned HTTP {code}: {detail}"); + } bail!( "sync endpoint returned HTTP {}: {}", code, @@ -148,11 +151,35 @@ impl SyncSink for HttpSink { } } +/// Renders a refused batch field into an actionable message. +/// +/// The endpoint refuses any record carrying a field it does not recognize. That +/// happens for two reasons worth telling apart: a collector sending data the +/// contract excludes, and a collector newer than the deployment it syncs to. +/// Only the field name crosses the wire, never its value. +fn unrecognized_field_detail(body: &str) -> Option { + let parsed = serde_json::from_str::(body).ok()?; + let first = parsed.get("rejected")?.as_array()?.first()?; + let field = first + .get("reason")? + .as_str()? + .strip_prefix("unknown_field:")?; + let kind = first + .get("kind") + .and_then(serde_json::Value::as_str) + .unwrap_or("record"); + Some(format!( + "endpoint does not recognize `{field}` on {kind}. \ + Deploy the API before upgrading collectors, or stop sending the field." + )) +} + fn validate_sync_ack(batch: &SyncBatch, ack: &SyncAck) -> Result<()> { let expected_ack_schema = match batch.schema_version.as_str() { SYNC_BATCH_V1_SCHEMA_VERSION => SYNC_ACK_V1_SCHEMA_VERSION, SYNC_BATCH_V2_SCHEMA_VERSION => SYNC_ACK_V2_SCHEMA_VERSION, SYNC_BATCH_V3_SCHEMA_VERSION => SYNC_ACK_V3_SCHEMA_VERSION, + SYNC_BATCH_V4_SCHEMA_VERSION => SYNC_ACK_V4_SCHEMA_VERSION, other => bail!("unsupported sync batch schema {other}"), }; if ack.schema_version != expected_ack_schema { @@ -235,6 +262,12 @@ fn validate_sync_ack(batch: &SyncBatch, ack: &SyncAck) -> Result<()> { ack.accepted.code_change_metrics, ack.duplicates.code_change_metrics, )?; + validate_sync_ack_counts( + "quota_cycle_contributions", + batch.quota_cycle_contributions.len() as u64, + ack.accepted.quota_cycle_contributions, + ack.duplicates.quota_cycle_contributions, + )?; Ok(()) } @@ -319,6 +352,31 @@ mod tests { use std::sync::mpsc; use tiny_http::{Header, Method, Response, Server}; + #[test] + fn unrecognized_field_refusals_name_the_field_and_the_remedy() { + let detail = unrecognized_field_detail( + r#"{"error":"invalid_sync_batch","rejected":[{"kind":"quota_cycle_contributions","id":"quota_cycle_aaaa","reason":"unknown_field:has_schedule_overlap"}]}"#, + ) + .expect("refusal detail"); + assert!(detail.contains("has_schedule_overlap")); + assert!(detail.contains("quota_cycle_contributions")); + assert!(detail.contains("Deploy the API before upgrading collectors")); + + let nested = unrecognized_field_detail( + r#"{"error":"invalid_sync_batch","rejected":[{"kind":"quota_cycle_contributions","id":null,"reason":"unknown_field:boundary_slices.working_directory"}]}"#, + ) + .expect("nested refusal detail"); + assert!(nested.contains("boundary_slices.working_directory")); + + // Anything that is not a field refusal falls back to the raw body. + assert!(unrecognized_field_detail(r#"{"error":"invalid_sync_batch"}"#).is_none()); + assert!(unrecognized_field_detail( + r#"{"rejected":[{"kind":"events","id":null,"reason":"raw_event_cloud_sync_disabled"}]}"# + ) + .is_none()); + assert!(unrecognized_field_detail("upstream timeout").is_none()); + } + fn empty_batch() -> SyncBatch { SyncBatch { schema_version: SYNC_BATCH_SCHEMA_VERSION.to_string(), @@ -333,6 +391,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: Utc::now(), } @@ -434,7 +493,7 @@ mod tests { let (auth, content_type, body) = rx.recv().expect("request body"); assert_eq!(auth.as_deref(), Some("Bearer token_123")); assert_eq!(content_type.as_deref(), Some("application/json")); - assert!(body.contains("\"schema_version\":\"sync_batch.v3\"")); + assert!(body.contains("\"schema_version\":\"sync_batch.v4\"")); assert!(body.contains("\"batch_id\":\"batch_1\"")); } @@ -533,8 +592,8 @@ mod tests { )) .expect("v1 ack"); - let error = validate_sync_ack(&batch, &ack).expect_err("v3 batch with v1 ack"); - assert!(error.to_string().contains("requires sync_ack.v3")); + let error = validate_sync_ack(&batch, &ack).expect_err("v4 batch with v1 ack"); + assert!(error.to_string().contains("requires sync_ack.v4")); batch.schema_version = SYNC_BATCH_V1_SCHEMA_VERSION.to_string(); ack.schema_version = SYNC_ACK_V2_SCHEMA_VERSION.to_string(); @@ -547,6 +606,10 @@ mod tests { batch.schema_version = SYNC_BATCH_V3_SCHEMA_VERSION.to_string(); ack.schema_version = SYNC_ACK_V3_SCHEMA_VERSION.to_string(); validate_sync_ack(&batch, &ack).expect("matching v3 schemas"); + + batch.schema_version = SYNC_BATCH_V4_SCHEMA_VERSION.to_string(); + ack.schema_version = SYNC_ACK_V4_SCHEMA_VERSION.to_string(); + validate_sync_ack(&batch, &ack).expect("matching v4 schemas"); } #[test] diff --git a/crates/statsai/src/main.rs b/crates/statsai/src/main.rs index 53f0434..26bdb2f 100644 --- a/crates/statsai/src/main.rs +++ b/crates/statsai/src/main.rs @@ -59,6 +59,7 @@ use statsai::{auth, default_device_id, default_store_path, service, snapshot}; const HTTP_ROLLUP_SUMMARIES_PER_BATCH: usize = 25; const HTTP_ROLLUP_METADATA_RECORDS_PER_BATCH: usize = 20; const HTTP_ROLLUP_CODE_CHANGE_METRICS_PER_BATCH: usize = 1_000; +const HTTP_ROLLUP_QUOTA_CYCLE_CONTRIBUTIONS_PER_BATCH: usize = 100; const HTTP_ROLLUP_D1_QUERY_BUDGET: usize = 45; const HTTP_ROLLUP_D1_QUERY_CHUNK_SIZE: usize = 90; const HTTP_ROLLUP_DAILY_ROLLUP_ROWS_PER_QUERY: usize = 7; @@ -922,7 +923,7 @@ enum StoreAdminSubcommand { #[derive(Debug, Subcommand)] enum SchemaSubcommand { - #[command(about = "Print the sync_batch.v3 JSON Schema")] + #[command(about = "Print the sync_batch.v4 JSON Schema")] SyncBatch, #[command(about = "Print the quota_window_sync_projection.v1 JSON Schema")] QuotaWindowProjection, @@ -4139,6 +4140,21 @@ fn build_sync_batch_with_identity_key( .iter() .map(|metric| metric.metric_id.clone()) .collect::>(); + let all_quota_cycle_contributions = + store.quota_cycle_contributions(&QuotaQuery::default(), device_id)?; + let quota_cycle_contributions = if command.full || state.is_none() { + all_quota_cycle_contributions.clone() + } else { + store.pending_quota_cycle_contributions_for_sync( + &command.sink, + target, + &all_quota_cycle_contributions, + )? + }; + let all_quota_cycle_contribution_ids = all_quota_cycle_contributions + .iter() + .map(|contribution| contribution.contribution_id.clone()) + .collect::>(); if payload_mode == SyncPayloadMode::Rollups { let label = rollup_mode_label(command); @@ -4178,6 +4194,7 @@ fn build_sync_batch_with_identity_key( .map(|summary| summary.summary_id.clone()) .collect(), code_change_metric_ids: all_code_change_metric_ids, + quota_cycle_contribution_ids: all_quota_cycle_contribution_ids, }; let failed_without_resume = state.as_ref().is_some_and(|state| { state.failure_count > 0 && state.pending_resume_batch_id.is_none() @@ -4240,6 +4257,7 @@ fn build_sync_batch_with_identity_key( task_buckets, task_verifications, code_change_metrics, + quota_cycle_contributions, authoritative_snapshot, created_at, }, @@ -4278,6 +4296,11 @@ fn record_rollup_sync_chunk_success( .map(|metric| metric.metric_id.clone()) .collect::>(), )?; + store.record_quota_cycle_contributions_synced( + sink, + target, + &batch.quota_cycle_contributions, + )?; snapshot::invalidate_dashboard_cache(); Ok(()) } @@ -4312,6 +4335,11 @@ fn record_sync_batch_success( .map(|metric| metric.metric_id.clone()) .collect::>(), )?; + store.record_quota_cycle_contributions_synced( + sink, + target, + &batch.quota_cycle_contributions, + )?; snapshot::invalidate_dashboard_cache(); Ok(()) } @@ -4630,9 +4658,11 @@ fn should_retry_http_rollup_chunk_after_error(chunk: &SyncBatch, error: &anyhow: || chunk.task_buckets.len() > 1 || chunk.task_verifications.len() > 1 || chunk.code_change_metrics.len() > 1 + || chunk.quota_cycle_contributions.len() > 1 || (!chunk.task_buckets.is_empty() && !chunk.task_verifications.is_empty()) || (http_rollup_metadata_count(chunk) > 0 && !chunk.summaries.is_empty()) || (!chunk.code_change_metrics.is_empty() && has_non_code_change_payload(chunk)) + || (!chunk.quota_cycle_contributions.is_empty() && has_non_quota_cycle_payload(chunk)) } fn http_rollup_retry_error_label(error: &anyhow::Error) -> &'static str { @@ -4711,6 +4741,7 @@ fn split_authoritative_snapshot( subscription_ids: Vec::new(), summary_ids: Vec::new(), code_change_metric_ids: Vec::new(), + quota_cycle_contribution_ids: Vec::new(), }; let mut parts = Vec::new(); let mut current = empty_part(); @@ -4735,6 +4766,10 @@ fn split_authoritative_snapshot( append_ids!(snapshot.subscription_ids, subscription_ids); append_ids!(snapshot.summary_ids, summary_ids); append_ids!(snapshot.code_change_metric_ids, code_change_metric_ids); + append_ids!( + snapshot.quota_cycle_contribution_ids, + quota_cycle_contribution_ids + ); if authoritative_snapshot_id_count(¤t) > 0 || parts.is_empty() { parts.push(current); } @@ -4753,6 +4788,7 @@ fn authoritative_snapshot_id_count(snapshot: &SyncAuthoritativeSnapshot) -> usiz + snapshot.subscription_ids.len() + snapshot.summary_ids.len() + snapshot.code_change_metric_ids.len() + + snapshot.quota_cycle_contribution_ids.len() } fn split_http_rollup_sync_batches_without_snapshot(batch: &SyncBatch) -> Vec { @@ -4763,11 +4799,14 @@ fn split_http_rollup_sync_batches_without_snapshot(batch: &SyncBatch) -> Vec 0 || !batch.summaries.is_empty() || !batch.code_change_metrics.is_empty(); + let has_rollup_payload = metadata_count > 0 + || !batch.summaries.is_empty() + || !batch.code_change_metrics.is_empty() + || !batch.quota_cycle_contributions.is_empty(); if !has_task_payload && batch.summaries.len() <= HTTP_ROLLUP_SUMMARIES_PER_BATCH && batch.code_change_metrics.len() <= HTTP_ROLLUP_CODE_CHANGE_METRICS_PER_BATCH + && batch.quota_cycle_contributions.len() <= HTTP_ROLLUP_QUOTA_CYCLE_CONTRIBUTIONS_PER_BATCH && metadata_count <= HTTP_ROLLUP_METADATA_RECORDS_PER_BATCH { return fit_http_rollup_batches_to_d1_budget(vec![batch.clone()]); @@ -4785,8 +4824,17 @@ fn split_http_rollup_sync_batches_without_snapshot(batch: &SyncBatch) -> Vec Vec Vec Vec { + if !batch.quota_cycle_contributions.is_empty() && has_non_quota_cycle_payload(batch) { + let mut without_quota = batch.clone(); + without_quota.quota_cycle_contributions.clear(); + let mut chunks = split_http_quota_cycle_contribution_chunks( + batch, + batch.quota_cycle_contributions.len(), + ); + chunks.extend(split_http_rollup_sync_batch_after_budget_error( + &without_quota, + )); + return chunks; + } + if batch.quota_cycle_contributions.len() > 1 { + return split_http_quota_cycle_contribution_chunks( + batch, + batch.quota_cycle_contributions.len().div_ceil(2), + ); + } if !batch.code_change_metrics.is_empty() && has_non_code_change_payload(batch) { let mut without_code_changes = batch.clone(); without_code_changes.code_change_metrics.clear(); @@ -4885,6 +4955,16 @@ fn has_non_code_change_payload(batch: &SyncBatch) -> bool { || !batch.summaries.is_empty() || !batch.task_buckets.is_empty() || !batch.task_verifications.is_empty() + || !batch.quota_cycle_contributions.is_empty() +} + +fn has_non_quota_cycle_payload(batch: &SyncBatch) -> bool { + http_rollup_metadata_count(batch) > 0 + || !batch.events.is_empty() + || !batch.summaries.is_empty() + || !batch.task_buckets.is_empty() + || !batch.task_verifications.is_empty() + || !batch.code_change_metrics.is_empty() } fn http_rollup_metadata_count(batch: &SyncBatch) -> usize { @@ -4943,6 +5023,22 @@ fn split_http_code_change_metric_chunks(batch: &SyncBatch, chunk_size: usize) -> .collect() } +fn split_http_quota_cycle_contribution_chunks( + batch: &SyncBatch, + chunk_size: usize, +) -> Vec { + batch + .quota_cycle_contributions + .chunks(chunk_size.max(1)) + .enumerate() + .map(|(index, contributions)| { + let mut chunk = empty_http_rollup_chunk(batch, &format!("quota_cycles_{}", index + 1)); + chunk.quota_cycle_contributions = contributions.to_vec(); + chunk + }) + .collect() +} + fn fit_http_rollup_batches_to_d1_budget(chunks: Vec) -> Vec { let mut fitted = Vec::new(); for chunk in chunks { @@ -5016,6 +5112,8 @@ fn estimate_http_rollup_d1_queries(batch: &SyncBatch) -> usize { // The backend batches metrics into one json_each upsert and one ownership // statement regardless of metric count. let code_change_metric_queries = usize::from(!batch.code_change_metrics.is_empty()) * 2; + let quota_cycle_contribution_queries = + usize::from(!batch.quota_cycle_contributions.is_empty()) * 2; let code_change_owner_metadata_refresh_queries = usize::from( batch.schema_version == SYNC_BATCH_SCHEMA_VERSION && batch @@ -5038,6 +5136,7 @@ fn estimate_http_rollup_d1_queries(batch: &SyncBatch) -> usize { + monthly_rollup_queries + dashboard_snapshot_queries + code_change_metric_queries + + quota_cycle_contribution_queries + code_change_owner_metadata_refresh_queries + estimate_http_rollup_task_queries(batch) + final_sync_bookkeeping_queries @@ -5410,6 +5509,7 @@ fn split_http_rollup_summary_chunks(batch: &SyncBatch, chunk_size: usize) -> Vec chunk.task_buckets.clear(); chunk.task_verifications.clear(); chunk.code_change_metrics.clear(); + chunk.quota_cycle_contributions.clear(); chunk.authoritative_snapshot = None; chunk }) @@ -5428,6 +5528,7 @@ fn empty_http_rollup_chunk(batch: &SyncBatch, suffix: &str) -> SyncBatch { chunk.task_buckets.clear(); chunk.task_verifications.clear(); chunk.code_change_metrics.clear(); + chunk.quota_cycle_contributions.clear(); chunk.authoritative_snapshot = None; chunk } @@ -6106,6 +6207,13 @@ fn quota(command: QuotaCommand, store: &Store, device_id: &str) -> Result<()> { format_u64(status.weekly_observations), status.weekly_sync_eligible_coverage_percent ); + let discarded = &status.discarded; + println!( + "reconstruction discarded: {} replayed observations, {} unused windows, {} bracketed schedules", + format_u64(discarded.replayed_observations), + format_u64(discarded.unused_windows), + format_u64(discarded.bracketed_schedules) + ); if status.unattributed_observations > 0 { eprintln!( "warning: historical quota observations remain unassigned; backdate a source connection with `statsai source connect --started-at ...`" @@ -14291,6 +14399,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -14339,6 +14448,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: Some(SyncAuthoritativeSnapshot { source_ids: vec![source.source_id.clone()], ..SyncAuthoritativeSnapshot::default() @@ -14390,6 +14500,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: Some(SyncAuthoritativeSnapshot { summary_ids, ..SyncAuthoritativeSnapshot::default() @@ -14509,6 +14620,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -14576,6 +14688,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -14668,6 +14781,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -15024,6 +15138,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -15369,6 +15484,7 @@ mod tests { task_buckets: test_task_only_sync_batch(now, 1, 0).task_buckets, task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -15453,6 +15569,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -15554,6 +15671,7 @@ mod tests { task_buckets: vec![], task_verifications: vec![], code_change_metrics: vec![], + quota_cycle_contributions: vec![], authoritative_snapshot: None, created_at: now, }; @@ -15701,6 +15819,7 @@ mod tests { task_buckets, task_verifications, code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: now, } @@ -15854,6 +15973,7 @@ mod tests { }], task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: now, } @@ -16012,6 +16132,7 @@ mod tests { task_buckets, task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: now, } @@ -18804,6 +18925,7 @@ mod tests { task_buckets: Vec::new(), task_verifications: Vec::new(), code_change_metrics: Vec::new(), + quota_cycle_contributions: Vec::new(), authoritative_snapshot: None, created_at: Utc .with_ymd_and_hms(2026, 6, 14, 13, 0, 0) diff --git a/docs/Observed Tools, MCP, and Skills Analytics TODO.md b/docs/Observed Tools, MCP, and Skills Analytics TODO.md new file mode 100644 index 0000000..3d2a1f0 --- /dev/null +++ b/docs/Observed Tools, MCP, and Skills Analytics TODO.md @@ -0,0 +1,70 @@ +# Observed Tools, MCP, and Skills Analytics + +## Summary + +- Collect activity automatically during normal `statsai scan`/daemon runs for Codex, Claude Code, OpenCode, and Grok Build; conversation archive collection remains optional. +- Read local records only—no OpenAI, Anthropic, OpenCode, or xAI account/API calls. Hosted data appears only through the existing StatsAI sync after explicit opt-in. +- Mirror the call-oriented approach used by [OpenAI Personal Analytics](https://help.openai.com/en/articles/20001478) and [OpenCode stats](https://github.com/anomalyco/opencode/blob/dev/packages/opencode/src/cli/cmd/stats.ts): rank observed calls/activations while keeping model tokens and cost separate. +- Add the experience at the bottom of `/dashboard/stats/`. Plugin identity is parent metadata for skills/MCPs, not a separate overlapping counter. +- Exclude Claude-style context-footprint history: `/context` reports what is loaded into the current context window, and that breakdown is not reliably persisted in local history. [Claude Code context documentation](https://code.claude.com/docs/en/debug-your-config) + +## Data and Metric Contracts + +- Add `ActivityInvocationV1` with a mutually exclusive `kind` of `tool`, `mcp`, or `skill`. Store a deterministic invocation ID, provider/source/account, UTC timestamp, exact display name, tool family, optional MCP server/tool, optional owning plugin, outcome, observed duration, detection evidence, and source-file hash. +- Use versioned tool families: `file-read`, `file-write`, `code-search`, `shell`, `web`, `browser`, `computer`, `agent`, `planning`, `media`, and `other`. Classification uses explicit aliases only; unknown names remain `other`. +- Never persist or sync arguments, results, commands, prompts, file paths, raw session/turn/call IDs, or assistant-prose inferences. +- Report only exact derived metrics: + - observed calls/skill uses, daily trend, share, active days, and last used; + - succeeded, failed, and unknown outcomes; + - success rate as `succeeded / (succeeded + failed)`; + - average duration from explicit duration fields or matched lifecycle timestamps for the same call ID, with sample count; + - direct cost only when the local record explicitly reports a per-call USD amount. Missing cost is `null`/“not reported,” never zero and never calculated from model tokens or public price tables. +- Attach detection coverage per provider and kind: + - Codex: native item-completion tool/MCP records are complete for that format; legacy wrapper records and catalog-resolved `SKILL.md` loads are partial. + - Claude Code: structured tool, MCP naming, and native Skill events are complete; outcome/duration coverage remains sample-based. + - OpenCode: structured tools and native `skill` parts are complete; MCP classification is partial unless configuration or record metadata identifies the server. + - Grok Build: structured tools are complete for supported log formats, MCP is explicit-record-only/partial, and skills are unavailable until Grok emits an explicit skill record. +- Count Codex skills only when an exact `SKILL.md` under a discovered skill catalog is loaded, or an explicit skill-reading tool identifies it. Random files named `SKILL.md`, catalog availability, and prose mentions do not count; UI labels this evidence as an observed load rather than claiming server-equivalent Skill calls. + +## Collector, Store, and Sync + +- Extend adapter scans and daemon persistence with activity records, bump each provider parser revision for a one-time historical backfill, and reconcile changed/deleted files transactionally with usage records. +- Store normalized invocations locally and materialize UTC daily activity rollups. Re-resolve provider-account ownership by invocation timestamp when assignments change, retiring obsolete rollup IDs. +- Add `statsai activity status [--provider …] [--kind tool|mcp|skill] [--json]`, returning compact totals, top entities, date range, sample coverage, parser coverage, and hosted-sync state. Scan preview/status also reports discovered activity without writing. +- Add independent sync preference `include_activity`, defaulting to `false`, with `sync --include-activity` and `--exclude-activity`. Enabling marks all activity rollups dirty for historical backfill; disabling schedules deletion of the device’s hosted activity on the next successful sync. +- Introduce: + - `activity_rollup.v1`: one row per device/source/account/day/entity, with readable names, counts, outcome totals, duration samples/sum, optional provider-reported direct micro-USD, and first/last observation times. + - `activity_coverage.v1`: one row per source/day/kind with coverage level, evidence kind, and parser revision, including zero-call days where the source format was observed. + - `sync_batch.v5`/`sync_ack.v5`, adding activity collections, counts, and `activity_rollup_ids`/`activity_coverage_ids` to authoritative snapshots. +- Exact skill, plugin, MCP-server, and custom-tool names sync only when activity is enabled. Project metadata is not part of activity v1. Turning activity off sends an authoritative empty activity set so the backend prunes readable names; sync v1–v4 never treat absent activity fields as authoritative. +- Deploy D1/API support for v5 before releasing the collector. Preserve v1–v4 ingestion compatibility, chunking/retry semantics, device ownership, remote reset, privacy inspection, and loopback diagnostic behavior. + +## Hosted API and Stats UI + +- Add authenticated `GET /api/dashboard/activity?range=<7d|30d|90d|all>&account=`. +- Return `dashboard_activity.v1` with: + - totals for observed tool calls, MCP calls, and skill uses; + - daily three-series activity; + - tool-family totals; + - up to 100 ranked rows per kind plus an exact “Other” aggregate; + - row fields for provider, family/exact name, MCP server or plugin parent, calls, share, active days, last use, outcome counts/coverage, duration average/sample count, nullable direct cost, cost coverage, and detection evidence; + - provider/kind coverage summaries. +- Store only bounded aggregate fields in D1, validate names and identifiers strictly, use indexed range/account/provider/entity queries, and add the endpoint to D1 query-budget checks. +- Append an “Agent activity” section after Model Summary on `/dashboard/stats/`: + - cards for Tools, MCP calls, Skill uses, and reported direct cost; + - a daily Tools/MCP/Skills chart; + - Tools, MCPs, and Skills tabs with sortable ranked tables; + - tool family alongside provider-exact name, `server · tool` for MCPs, and `plugin · skill` where ownership is known; + - shared account/provider/time filters, top-ten collapse with “Show all,” coverage badges, partial-data explanations, and an opt-in empty state showing `statsai sync --sink http --include-activity`. +- If no source reports direct per-call cost, hide monetary ranking and show “Direct cost is not separately reported”; existing model cost above remains unchanged. +- Keep activity out of public profiles, projects, team analytics, leaderboards, and task views in v1. + +## Test and Rollout Plan + +- Adapter fixtures cover every provider, parallel calls, native skills, catalog skill loads, MCP parsing, legacy formats, missing results, failures, duration samples, duplicates, malformed records, and unsupported coverage. +- Store tests cover idempotent rescans, automatic backfill, cache revision upgrades, source-file removal, account reassignment, preview no-writes, daemon parity, and rollup conservation. +- Privacy tests prove no arguments, outputs, commands, paths, session IDs, or call IDs enter activity sync payloads. +- Sync/API tests cover v1–v5 compatibility, enable/backfill, disable/prune, chunk ACK counts, retry idempotency, authoritative fragments, device ownership, remote reset, validation bounds, account/provider/range filters, nullable cost, and query budgets. +- UI tests cover hook normalization, tabs, grouping, sorting, shared filters, “Other,” empty/error/loading states, partial coverage, unknown outcomes/durations, and missing versus explicit-zero cost. +- Verify with relevant Rust workspace tests and checks, `apps/api` unit/type/D1-budget tests, and `ui` test/lint/build. +- Implement on top of the current v4 quota branch without overwriting its existing dirty quota/daemon changes or the API’s unrelated untracked documentation files. diff --git a/docs/quota-sync.md b/docs/quota-sync.md index 8eaedc3..1abcbe8 100644 --- a/docs/quota-sync.md +++ b/docs/quota-sync.md @@ -1,41 +1,104 @@ -# Quota window projection contract +# Quota cycle contribution contract -StatsAI stores provider quota observations locally and reconstructs reset windows without adding -them to `SyncBatch.v3`. The standalone hosted-facing contract is available from the CLI: +StatsAI stores provider quota observations locally and reconstructs reset +windows before sync. Hosted quota history travels inside `sync_batch.v4` as +`quota_cycle_contribution.v1` records. The standalone +`quota_window_sync_projection.v1` export remains available for diagnostics and +is not part of hosted sync. ```sh +statsai schema sync-batch statsai schema quota-window-projection ``` -`quota_window_sync_projection.v1` is a device contribution. Its `projection_id` is deterministic -for the device, provider account, limit identity, duration, and cluster anchor point. It is not a -logical account-window identifier. Projections omit source paths, local source/account row -identities, provider payloads, usage tokens, and cost. +A contribution is a deterministic device record for one attributed cycle. Its +`contribution_id` is derived from the device, provider account, limit identity, +nominal duration, and the first reconstructed change-point fingerprint. It is +not a logical account-cycle identifier. The authenticated device identity on +the wire is supplied by the backend token; contributions themselves do not +carry a `device_id`. -Only attributed windows lasting at least 10,080 minutes are emitted by: +Each hosted contribution contains only: -```sh -statsai quota export --level sync-windows --format jsonl -``` +- the deterministic contribution ID +- provider and provider-account ID +- optional provider limit ID and nominal duration +- the representative reset timestamp +- whether local reconstruction saw another cycle in the same scope overlapping + this one (`has_schedule_overlap`) +- UTC daily envelopes with timestamped first, last, minimum, and maximum percentages +- exact usage slices for partial UTC days at cycle and observed schedule-transition boundaries + +Usage slices contain token-category totals and estimated API-equivalent cost in +integer micro-USD. Contributions omit raw events, event hashes, paths, source +IDs, quota payloads, plans, credits, provider slots, fingerprints, and sample +counts. + +Codex currently emits only attributed weekly windows (`window_minutes = 10080`). +Five-hour, monthly, and unattributed cycles stay local. Reset timestamps may +shift, and consecutive cycles need not be exactly seven days apart. -The command warns when locally observed weekly history is not fully attributed. +A Codex weekly cycle starts lazily at the first request after the previous +reset, not at the previous reset itself. An idle stretch therefore leaves a gap +between cycles, while consecutive cycles are never closer than the nominal +duration unless the earlier one was reset ahead of schedule — banked by the user +or granted server-side. Overlapping schedules are consequently expected, and +`has_schedule_overlap` records that a single device reconstructed the overlap +from its own observations. -Before attribution, local window reconstruction is partitioned by stable source ID so evidence -from separate local provider installations cannot be blended. That source partition is local-only: -once evidence is attributed, account identity becomes the reconstruction scope, and no source ID is -included in the sync projection. +A reset is a handover: once a window resets, the provider reports the new +schedule and never returns to the old one. Local reconstruction therefore +discards any schedule that another schedule brackets — reported before it +appeared and again after it vanished — once that other schedule has passed +everything the bracketed one ever claimed. During the July 2026 tier migration +Codex answered scattered turns with a blank snapshot, near-zero usage against a +fresh weekly reset, interleaved with the live schedule for up to 34 hours; each +such run would otherwise become a cycle that takes days away from the cycle +actually running. A genuine early reset leads from the moment it starts and the +schedule it replaced stops being reported, so it can never be bracketed. + +Complete UTC days are not copied into the contribution. The backend joins +existing daily summaries for those days. Boundary slices exist only to avoid +inaccurate proration at cycle and schedule-transition edges. Usage observed on +different devices is additive; hosted sync does not deduplicate usage events +across devices. + +Only attributed cycles are synced. Before attribution, local reconstruction +is partitioned by stable source ID so evidence from separate local provider +installations cannot be blended. ## Server merge semantics -1. Group contributions by stable provider account, provider, limit identity, and duration. -2. Cluster reset epochs whose raw reset values are within five minutes. This tolerance covers - provider recomputation drift; it is not a device clock-skew correction. -3. Deduplicate change points by `point_fingerprint`. -4. Preserve the chronological percentage sequence. Do not sum percentages and do not replace the - sequence with a maximum observed percentage. -5. Retain the logical window while at least one device projection contributes it. -6. Join the merged interval to the existing server-deduplicated usage rollup for the provider - account. Never derive or sum token/cost values from quota projections. - -Adding this projection to a future hosted wire payload is a separate capability and versioning -decision; this contract does not change `SyncBatch.v3`. +1. Group contributions by canonical provider account, provider, limit identity, + and nominal duration. +2. Cluster reset candidates only when their total spread is at most five + minutes. This tolerance covers provider recomputation drift; it is not a + device clock-skew correction. +3. Use the median reset as the representative reset. +4. Drop daily envelopes whose first or last observation is more than one hour + after that contribution's representative reset. Provider recomputation lag of + about 47 minutes is genuine; replay stamps weeks later are not. Already-stored + rows are filtered on read, and ingest strips the envelopes rather than + rejecting the batch, so devices need not re-sync. +5. Merge remaining daily envelopes using earliest first, latest last, lowest + minimum, and highest maximum observations, and derive contributing-device + counts. +6. Never synthesize a zero-percent start. A device first observed at 80% joins + the existing cycle identified by its reset; another device may supply earlier + evidence. +7. Preserve cycles observed only by another device. +8. For overlapping schedules, use the first observation of the newer schedule as + the transition boundary, clamped to the overlap. Do not assign usage to both + cycles. When a device contributed to both cycles and flagged + `has_schedule_overlap` on each, the overlap is an expected early reset and + both cycles stay complete. An overlap no single device reconstructed is + reported as a conflict, because it can only arise from cross-device records + that failed to cluster or from two provider accounts resolving to one + identity. +9. Combine full UTC-day summaries with exact boundary slices. If required + boundary evidence is absent, return the available value as partial rather + than prorating it. + +A logical cycle remains while any device still contributes it. Stale device +contributions are retired through the same authoritative-snapshot ownership +used by summaries and code-change metrics. diff --git a/docs/sync-contract.md b/docs/sync-contract.md index 38fda13..d50c147 100644 --- a/docs/sync-contract.md +++ b/docs/sync-contract.md @@ -7,6 +7,9 @@ marker for deletion reconciliation. `sync_batch.v3` adds privacy-safe numeric code-change metrics and makes their acknowledgement counts and authoritative snapshot IDs part of the versioned contract. Versions 1 and 2 never treat those new fields as authoritative. +`sync_batch.v4` adds provider-neutral quota-cycle contributions and makes their +acknowledgement counts and authoritative snapshot IDs part of the versioned +contract. Versions 1–3 never treat those new fields as authoritative. The collector owns local scanning, normalization, idempotent local storage, and privacy scrubbing. The backend owns authentication, validation, deduplication, rollups, and user-facing queries. The production path sends sanitized batches to @@ -239,12 +242,15 @@ hostnames, including `localhost`, are rejected. The daemon still supports `/v1/sync/batches` for loopback-only diagnostics, but rejects batches containing `authoritative_snapshot` because it does not stage -device ownership or reconcile deletions. `/api/sync/batches` is the production -contract. A compatible backend should: +device ownership or reconcile deletions. It rejects batches carrying +`quota_cycle_contributions` for the same reason: a local store keeps quota +observations and derives its own cycles from them, so acknowledging another +device's cycles would tell the sender they had been stored when they had not. +`/api/sync/batches` is the production contract. A compatible backend should: - require an authenticated device access token - accept `Authorization: Bearer ` from stored auth, `--auth-token`, or `STATSAI_SYNC_TOKEN` -- validate the request body against `sync_batch.v1`, `sync_batch.v2`, and `sync_batch.v3` +- validate the request body against `sync_batch.v1`, `sync_batch.v2`, `sync_batch.v3`, and `sync_batch.v4` - reject unsupported `schema_version` values - deduplicate sources, accounts, source-account assignments, subscriptions, and summaries by their IDs when server-side deduplication is needed - treat collector IDs as stable client-provided IDs, not database primary keys exposed to users @@ -252,7 +258,8 @@ contract. A compatible backend should: - atomically replace accepted task bucket snapshots per `(user, device, project_bucket)` - treat the ordered `authoritative_snapshot` fragments sharing one `snapshot_id` as the complete set of metadata and summary IDs owned by the authenticated - device; v3 fragments also carry code-change metric IDs; each fragment carries + device; v3 fragments also carry code-change metric IDs; v4 fragments also + carry quota-cycle contribution IDs; each fragment carries zero-based `part_index` and a common `part_count`, with at most 200 IDs across its ID arrays - stage snapshot ownership without pruning until the final in-order fragment; @@ -311,14 +318,69 @@ run immediately. `task_verifications` counters under both `accepted` and `duplicates`. `sync_batch.v3` returns `sync_ack.v3`, which additionally adds the `code_change_metrics` counter. +`sync_batch.v4` returns `sync_ack.v4`, which additionally adds the +`quota_cycle_contributions` counter. Collectors require the acknowledgement version to match the submitted batch version exactly; a v1 acknowledgement cannot successfully acknowledge a v2 -batch, and a v2 acknowledgement cannot acknowledge a v3 batch. +batch, a v2 acknowledgement cannot acknowledge a v3 batch, and a v3 +acknowledgement cannot acknowledge a v4 batch. The current loopback daemon returns this shape and reports duplicate events when the existing store already has the semantic event. Source, account, source-account assignment, subscription, and summary upserts are currently -reported as accepted writes. +reported as accepted writes. Its `quota_cycle_contributions` counter is +therefore always zero: a batch carrying any is refused outright rather than +acknowledged, since the daemon has nowhere to store them. + +## Evolving the Contract + +Two kinds of addition behave very differently, and the difference decides the +release order. + +**A new top-level collection is backward compatible.** Batch collections carry +`#[serde(default, skip_serializing_if = "Vec::is_empty")]`, and the backend +reads named fields without a top-level allowlist, so a deployment that predates +the collection ignores it rather than failing. That is not silent data loss: +collectors require `accepted + duplicates == submitted` for every collection +they send, so an endpoint that ignores one reports zero against a non-zero +submission and the sync fails loudly with a count mismatch. Local sync state +stays dirty and the records are resent. Adding a collection therefore needs no +schema-version bump. + +**A new field on an existing record is not.** Every synced record is checked +against a closed set of permitted keys, and an unrecognized key refuses the +whole batch with `400 invalid_sync_batch`. This is deliberate: an unexpected +key may carry a path, address, or message that this contract excludes, and the +refusal guarantees none of it is stored. The consequence is a hard release +order: + +> **Deploy the backend before releasing a collector that adds a record field.** +> Until it is deployed, upgraded collectors cannot sync anything — not just the +> affected collection. + +To keep the two causes distinguishable, the refusal names the offending field: + +```json +{ + "error": "invalid_sync_batch", + "rejected": [ + { + "kind": "quota_cycle_contributions", + "id": "quota_cycle_0f2c…", + "reason": "unknown_field:has_schedule_overlap" + } + ] +} +``` + +Nested records report a dotted path, such as +`unknown_field:boundary_slices.working_directory`. Only the field name is +returned; its value is never echoed. Collectors render this as +`endpoint does not recognize \`\` on ` rather than an opaque +HTTP error. + +Malformed values inside a *known* field remain a plain `400` with no `rejected` +detail — that is a client defect, not version skew. ## Local Sync State @@ -331,7 +393,8 @@ metadata. Full HTTP rollup syncs send their authoritative snapshot as the final logical chunk. The marker lists all current source, provider-account, -source-account-assignment, subscription, summary, and code-change metric IDs, +source-account-assignment, subscription, summary, code-change metric, and +quota-cycle contribution IDs, including empty lists. The backend tracks ownership per authenticated device and keeps device-local IDs separate from server-canonical IDs so account alias reconciliation cannot delete @@ -345,7 +408,8 @@ snapshot are retained through their active ownership mapping; omitted unowned rows are pruned. Legacy rows from other devices, and canonical rows still owned by any device, are preserved. -The HTTP sink parses `sync_ack.v1`, `sync_ack.v2`, and `sync_ack.v3` before +The HTTP sink parses `sync_ack.v1`, `sync_ack.v2`, `sync_ack.v3`, and +`sync_ack.v4` before updating local state. File and stdout sinks update state after their local write succeeds. @@ -380,8 +444,9 @@ Auth token precedence for sync is: The Worker rejects raw event cloud sync by default and accepts sanitized daily summary rollups plus metadata, along with hosted task snapshots and hosted task -verification actions for `sync_batch.v2` and `sync_batch.v3`, plus code-change -metrics for `sync_batch.v3`. The collector now prepares those +verification actions for `sync_batch.v2` and later, plus code-change +metrics for `sync_batch.v3` and later, plus quota-cycle contributions for +`sync_batch.v4`. The collector now prepares those daily rollups before HTTP sync, so a normal Cloudflare sync can populate the dashboard without shipping raw events. Repeated batches are idempotent by stable IDs.