diff --git a/README.md b/README.md index 2e2bc9f..4a5995f 100644 --- a/README.md +++ b/README.md @@ -178,8 +178,22 @@ stream plus the latest published read-model. MCP-only readers pull the semantic index and compact analysis projection. With `--athena-workgroup`, their trace tools query time/stream-pruned raw event rows directly and do not download `trace.json` or mirror raw chunks. -A bounded stream registry and per-stream local key cursors avoid relisting -historical chunks on each local read. The TUI builds unpublished event deltas in the background; +A bounded stream registry, per-day object watermarks, and local key cursors +avoid relisting unchanged historical chunks. A reader scans pre-watermark +history once, then lists only days whose published high-water key advanced. +New uploads use each event's UTC day, while +`event-partitions/.json` records the event-time range of every physical +day and `event-partitions//track..json` records each immutable +object's range. Athena combines projected stream/day partitions with its hidden +`$path` column, so a wide legacy capture day can prune unrelated objects. +Once a stream index exists, readers include its legacy days and objects missing +from the object index conservatively. A stream with no partition metadata falls +back to physical days overlapping the requested window; initialize or backfill +the metadata before relying on delayed historical event-time lookups. No raw +object migration is required. A one-time metadata-only range scan is recommended +before broad historical MCP use; it writes only these small indexes and does +not copy, rewrite, or delete JSONL. The TUI builds +unpublished event deltas in the background; `synty build` does the same explicitly, while `search` warns if raw events are newer than the published index. One tokened machine scrapes GitHub for everyone. @@ -238,12 +252,17 @@ MCP pulls the published semantic index and compact analysis projection before serving and refreshes them on a background thread; it never mirrors the raw event lake. In Athena mode it deliberately omits the legacy `trace.json` blob. Each trace request is a read-only `SELECT`, partition-pruned by stream and day, -limited to seven days, 50,000 events, 64 MiB of returned envelopes, a 50-second -query timeout, and the workgroup's 20 GiB scan cutoff. Limit hits fail closed -and ask the caller for a narrower time/machine/source/operation filter. +limited to seven days, 50,000 events, 64 MiB of returned envelopes, one shared +45-second Athena budget across all queries needed by the request, 1,000 raw +object paths per query, and the workgroup's 20 GiB scan cutoff. Success, error, +timeout, and limit outcomes use the standard metrics block. Limit hits fail +closed and ask the caller for a narrower time/machine/source/operation filter +or a metadata-only object-range backfill. `/health` reports transport liveness and `/ready` waits for the semantic index -and analysis projection, plus both dispatchers (`trace.json` is required only -in local-projection mode). Analysis tools are +and analysis projection, a compatible remote read-model format, plus both +dispatchers (`trace.json` is required only in local-projection mode). Status +and both health endpoints report the newest indexed raw-event timestamp, the +published-model timestamp/format, and the last bucket metadata check. Analysis tools are serialized on a one-slot dispatcher so concurrent first loads cannot multiply memory. HTTP work is bounded by separate semantic and analysis queues, a 120-second response deadline, 32 in-flight requests, 64 live connections, diff --git a/deploy/aws/mcp-reader.yaml b/deploy/aws/mcp-reader.yaml index 74df895..9d378d2 100644 --- a/deploy/aws/mcp-reader.yaml +++ b/deploy/aws/mcp-reader.yaml @@ -30,7 +30,6 @@ Parameters: GlueTable: Type: String Default: raw_events - Resources: ReaderRole: Type: AWS::IAM::Role diff --git a/docs/design.md b/docs/design.md index d1d75fe..828b672 100644 --- a/docs/design.md +++ b/docs/design.md @@ -46,7 +46,7 @@ flowchart LR B --> P["Published next-plaid read model"] E --> B P -->|"download + mmap"| MCP["Read-only MCP"] - S3 -->|"external table; no migration"| G["Glue Data Catalog"] + S3 -->|"external table; no migration"| G["Glue raw_events"] G --> A["Bounded Athena SELECT"] A -->|"bounded raw envelopes"| F["Rust trace fold"] F --> MCP @@ -62,11 +62,13 @@ jobs. The MCP pod does not need a trace projection job or `trace.json` on its volume. Local CLI/TUI operation stays self-contained and can use the local projection offline. -The raw-table overlay is the zero-copy bootstrap, not a columnar rewrite: -Athena still scans the selected JSONL object bytes. If daily raw volume reaches -the workgroup cutoff, a follow-up can write a separate immutable Parquet -projection partitioned for session lookup while leaving the authoritative raw -prefix untouched. This path deliberately creates no compaction job or migration. +The raw-table overlay is a zero-copy query path, not a columnar rewrite: +Athena still scans the selected JSONL object bytes. A compact per-stream +partition-range index maps event-time windows onto physical days, including +legacy capture-day chunks; per-day object-range indexes then prune immutable +files through Athena's hidden `$path` column. Per-day object high-water keys let +local readers list only partitions that changed after a one-time compatibility +scan. Neither metadata path rewrites the authoritative objects. ## Engine @@ -217,12 +219,21 @@ runs on CI or a server without a developer machine. artifacts in the background; it does not mirror the raw event lake. With `--athena-workgroup`, trace calls issue only bounded `SELECT` statements over the existing S3 event chunks and fold the returned rows in Rust. The backend - discovers injected stream partitions from `event-streams/`, caps the time - window at seven days, rows at 50,000, returned bytes at 64 MiB, query time at - 50 seconds, and relies on a workgroup scan cutoff as the final cost guard. + discovers injected stream partitions from `event-streams/`, resolves + event-time windows through `event-partitions/.json`, caps the time + window at seven days, rows at 50,000, returned bytes at 64 MiB, and all + Athena work for one request at a shared 45-second budget. Queries include at + most 1,000 exact raw-object `$path` values. A workgroup scan cutoff is the + final cost guard. Indexed streams include declared legacy days and physically + present objects missing from per-object metadata conservatively; a completely + unindexed stream falls back to the requested physical-day span. `/health` remains a liveness check, while `/ready` requires the semantic - index, compact analysis projection, and both dispatchers; local-projection - mode additionally requires `trace.json`. Analysis calls use a + index, compact analysis projection, a compatible bucket read-model format, + and both dispatchers; local-projection mode additionally requires + `trace.json`. Health and status expose bucket raw/model freshness; the + unauthenticated HTTP endpoints return only a generic metadata-error indicator, + while detailed provider diagnostics stay in server logs and protected status. + Analysis calls use a serialized one-slot dispatcher so concurrent first loads cannot multiply memory or block semantic search. Each dispatcher has a bounded queue; HTTP clients have a 120-second response deadline and a per-client @@ -327,12 +338,19 @@ bucket and drops into the viewer, so a paste goes from nothing to tracking. ## Storage layout (bucket) ```text -events//chunks//.jsonl +events//chunks/track./.jsonl immutable append deltas (source of truth); stream = edge--, so many trackers' files coexist without collision -event-streams/ immutable bounded discovery registry; - readers continue each stream by key cursor +event-streams/ immutable bounded discovery registry +event-partitions/.json physical day → complete event-time range + plus day → highest immutable object key; + legacy/unindexed days remain unconditional + query candidates (metadata-only; mutable) +event-partitions//track..json + immutable object → event-time range; + Athena uses exact $path pruning; + readers continue each day by key cursor members//activation.json immutable init access marker (no session data) embeddings//.emb content-addressed f16 vectors (write-once) summaries//-.json per-(unit, input-hash) LLM summaries @@ -361,9 +379,13 @@ fleet model is **no designated builder**: every tracker pushes events; whoever opens a local viewer pulls all raw streams and the published read-model, then contributes a build. MCP-only readers pull semantic/analysis artifacts without bucket write access or a raw-history mirror; an S3 reader can query trace rows -through the read-only Glue/Athena overlay. Local commands can inspect and -rebuild from every machine, while semantic results cover the latest published -build and warn when newer raw chunks are pending. +through the read-only Glue/Athena overlay. The partition-range index makes +delayed event-time windows queryable without moving JSONL: new writers publish +by event UTC day, while days created by old writers remain conservative until +an optional metadata-only range backfill proves their exact day and object +coverage. Local +commands can inspect and rebuild from every machine, while semantic results +cover the latest published build and warn when newer raw chunks are pending. Write-once stores are the collaboration primitive: a viewer encodes and summarizes only what no other machine has (pending lists shuffle per machine, so concurrent viewers split the work). The lease only prevents duplicate index diff --git a/src/event_partitions.rs b/src/event_partitions.rs new file mode 100644 index 0000000..7b06e17 --- /dev/null +++ b/src/event_partitions.rs @@ -0,0 +1,559 @@ +// Event-time coverage for physical raw-event partitions. Trackers publish one +// small mutable index per owned stream; Athena readers use it to choose the +// capture-day directories that can contain a requested event-time window. +// Missing or legacy coverage stays conservative: include the physical day +// rather than risk a false negative. + +use crate::bucket::Bucket; +use anyhow::{Result, anyhow}; +use chrono::{DateTime, NaiveDate, Utc}; +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, BTreeSet}; + +pub(crate) const PREFIX: &str = "event-partitions"; +const FORMAT: u32 = 1; +const WRITER_LAYOUT: u32 = 1; + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub(crate) struct EventTimeRange { + pub min_ts: String, + pub max_ts: String, +} + +impl EventTimeRange { + fn record(&mut self, ts: &str) { + let Some(timestamp) = parse_time(ts) else { + return; + }; + if self.min_ts.is_empty() + || parse_time(&self.min_ts).is_some_and(|current| timestamp < current) + { + self.min_ts = ts.to_string(); + } + if self.max_ts.is_empty() + || parse_time(&self.max_ts).is_some_and(|current| timestamp > current) + { + self.max_ts = ts.to_string(); + } + } + + #[cfg_attr(not(feature = "athena"), allow(dead_code))] + fn overlaps(&self, since: DateTime, until: DateTime) -> bool { + let Some(min) = parse_time(&self.min_ts) else { + return true; + }; + let Some(max) = parse_time(&self.max_ts) else { + return true; + }; + max >= since && min < until + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct EventPartitionIndex { + #[serde(default = "format")] + pub format: u32, + pub stream: String, + /// Physical days whose exact event-time coverage has not been initialized. + /// Readers always include them, preserving correctness for old writers. + #[serde(default)] + pub legacy_days: BTreeSet, + /// Physical day → inclusive event-time range for every object in that day. + #[serde(default)] + pub partitions: BTreeMap, + /// Highest immutable object key published for each event day. Readers + /// compare this watermark with their per-day cursor and list only days + /// that changed, including a newly published older day. + #[serde(default)] + pub object_cursors: BTreeMap, + /// New writers maintain `object_cursors`. Missing means an older writer + /// may still append outside those watermarks and needs the compatibility + /// stream scan. + #[serde(default)] + pub writer_layout: u32, + #[serde(default)] + pub updated_at: String, +} + +fn format() -> u32 { + FORMAT +} + +impl EventPartitionIndex { + fn empty(stream: &str) -> Self { + Self { + format: FORMAT, + stream: stream.to_string(), + legacy_days: BTreeSet::new(), + partitions: BTreeMap::new(), + object_cursors: BTreeMap::new(), + writer_layout: 0, + updated_at: String::new(), + } + } + + #[cfg_attr(not(feature = "athena"), allow(dead_code))] + pub(crate) fn candidate_days( + &self, + since: DateTime, + until: DateTime, + ) -> BTreeSet { + let mut days = self.legacy_days.clone(); + days.extend( + self.partitions + .iter() + .filter(|(_, range)| range.overlaps(since, until)) + .map(|(day, _)| day.clone()), + ); + days + } + + pub(crate) fn newest_event(&self) -> Option { + self.partitions + .values() + .filter_map(|range| parse_time(&range.max_ts)) + .max() + .map(|timestamp| timestamp.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true)) + } + + pub(crate) fn record_object(&mut self, day: &str, key: &str) { + let cursor = self.object_cursors.entry(day.to_string()).or_default(); + if key > cursor.as_str() { + *cursor = key.to_string(); + } + } + + /// Mark a historical physical day as completely indexed. This is the + /// metadata-only backfill seam: callers may derive the range with Athena, + /// then remove the conservative legacy marker without moving raw objects. + #[cfg(test)] + fn complete_legacy_day(&mut self, day: &str, min_ts: &str, max_ts: &str) { + self.legacy_days.remove(day); + self.partitions.insert( + day.to_string(), + EventTimeRange { + min_ts: min_ts.to_string(), + max_ts: max_ts.to_string(), + }, + ); + } +} + +pub(crate) fn key(stream: &str) -> String { + format!("{PREFIX}/{stream}.json") +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct EventObjectIndex { + #[serde(default = "format")] + pub format: u32, + pub stream: String, + pub day: String, + #[serde(default)] + pub objects: BTreeMap, + #[serde(default)] + pub updated_at: String, +} + +impl EventObjectIndex { + fn empty(stream: &str, day: &str) -> Self { + Self { + format: FORMAT, + stream: stream.to_string(), + day: day.to_string(), + objects: BTreeMap::new(), + updated_at: String::new(), + } + } + + #[cfg_attr(not(feature = "athena"), allow(dead_code))] + pub(crate) fn candidate_objects( + &self, + since: DateTime, + until: DateTime, + ) -> BTreeSet { + self.objects + .iter() + .filter(|(_, range)| range.overlaps(since, until)) + .map(|(key, _)| key.clone()) + .collect() + } +} + +pub(crate) fn object_key(stream: &str, day: &str) -> String { + format!("{PREFIX}/{stream}/track.{day}.json") +} + +pub(crate) fn load_objects( + bucket: &dyn Bucket, + stream: &str, + day: &str, +) -> Result> { + let Some(raw) = bucket.get(&object_key(stream, day))? else { + return Ok(None); + }; + let index: EventObjectIndex = serde_json::from_slice(&raw) + .map_err(|error| anyhow!("invalid event object index for {stream}/{day}: {error}"))?; + anyhow::ensure!( + index.format == FORMAT && index.stream == stream && index.day == day, + "event object index identity mismatch for {stream}/{day}" + ); + Ok(Some(index)) +} + +pub(crate) fn load_objects_or_empty( + bucket: &dyn Bucket, + stream: &str, + day: &str, +) -> Result { + Ok(load_objects(bucket, stream, day)?.unwrap_or_else(|| EventObjectIndex::empty(stream, day))) +} + +pub(crate) fn save_objects(bucket: &dyn Bucket, index: &EventObjectIndex) -> Result<()> { + let mut index = index.clone(); + index.updated_at = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + bucket.put( + &object_key(&index.stream, &index.day), + &serde_json::to_vec(&index)?, + ) +} + +/// Record exact event-time coverage for one immutable JSONL object. Malformed +/// timestamps leave the object unindexed; readers discover it from the +/// physical listing and include it conservatively. +pub(crate) fn record_object(index: &mut EventObjectIndex, key: &str, raw: &[u8]) { + let mut range = EventTimeRange::default(); + let mut records = 0usize; + for line in raw.split(|byte| *byte == b'\n') { + if line.iter().all(|byte| byte.is_ascii_whitespace()) { + continue; + } + let timestamp = serde_json::from_slice::(line) + .ok() + .and_then(|value| value["ts"].as_str().and_then(parse_time)); + let Some(timestamp) = timestamp else { + return; + }; + range.record(×tamp.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true)); + records += 1; + } + if records > 0 { + index.objects.insert(key.to_string(), range); + } +} + +pub(crate) fn load(bucket: &dyn Bucket, stream: &str) -> Result> { + let Some(raw) = bucket.get(&key(stream))? else { + return Ok(None); + }; + let index: EventPartitionIndex = serde_json::from_slice(&raw) + .map_err(|error| anyhow!("invalid event partition index for {stream}: {error}"))?; + anyhow::ensure!( + index.format == FORMAT && index.stream == stream, + "event partition index identity mismatch for {stream}" + ); + Ok(Some(index)) +} + +/// Initialize before publishing event-day chunks. Every already-present day is +/// legacy and therefore always queried until an operator supplies its complete +/// min/max range. The object is written first, so a crash cannot make a new +/// layout silently look complete. +pub(crate) fn load_or_initialize(bucket: &dyn Bucket, stream: &str) -> Result { + if let Some(index) = load(bucket, stream)? { + return Ok(index); + } + let mut index = EventPartitionIndex::empty(stream); + for object in bucket.list(&format!("events/{stream}/chunks/"))? { + if let Some(day) = day_from_event_key(&object) { + index.legacy_days.insert(day.to_string()); + } + } + save(bucket, &index)?; + Ok(index) +} + +pub(crate) fn save(bucket: &dyn Bucket, index: &EventPartitionIndex) -> Result<()> { + let mut index = index.clone(); + index.writer_layout = WRITER_LAYOUT; + index.updated_at = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + bucket.put(&key(&index.stream), &serde_json::to_vec(&index)?) +} + +/// Newest indexed raw-event timestamp across the bucket. This reads only the +/// small per-stream metadata objects, never the JSONL event bodies. +pub(crate) fn bucket_newest_event(bucket: &dyn Bucket) -> Result> { + let mut newest: Option> = None; + for key in bucket.list(crate::sync::EVENT_STREAMS)? { + let Some(stream) = key + .strip_prefix(&format!("{}/", crate::sync::EVENT_STREAMS)) + .filter(|stream| !stream.is_empty() && !stream.contains('/')) + else { + continue; + }; + let Some(index) = load(bucket, stream)? else { + continue; + }; + if let Some(timestamp) = index.newest_event().as_deref().and_then(parse_time) + && newest.is_none_or(|current| timestamp > current) + { + newest = Some(timestamp); + } + } + Ok(newest + .map(|timestamp| timestamp.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true))) +} + +/// Split complete JSONL lines by each event's UTC day and update the exact +/// range for those physical partitions. Unknown timestamps remain readable in +/// the source file's day, which is marked legacy so queries include it. +pub(crate) fn partition_lines( + index: &mut EventPartitionIndex, + raw: &[u8], + fallback_day: &str, +) -> BTreeMap> { + let mut by_day: BTreeMap> = BTreeMap::new(); + for line in raw.split_inclusive(|byte| *byte == b'\n') { + if line.iter().all(|byte| byte.is_ascii_whitespace()) { + continue; + } + let timestamp = serde_json::from_slice::(line) + .ok() + .and_then(|value| value["ts"].as_str().map(str::to_string)); + let (day, timestamp) = timestamp + .as_deref() + .and_then(parse_time) + .map(|timestamp| { + ( + timestamp.format("%Y-%m-%d").to_string(), + Some(timestamp.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true)), + ) + }) + .unwrap_or_else(|| (fallback_day.to_string(), None)); + by_day + .entry(day.clone()) + .or_default() + .extend_from_slice(line); + if let Some(timestamp) = timestamp.as_deref() { + index.partitions.entry(day).or_default().record(timestamp); + } else { + index.legacy_days.insert(day); + } + } + by_day +} + +pub(crate) fn day_from_event_key(key: &str) -> Option<&str> { + let rest = key.split_once("/chunks/track.")?.1; + let day = rest.split('/').next()?; + valid_day(day).then_some(day) +} + +pub(crate) fn day_from_track_file(file: &str) -> Option<&str> { + let day = file.strip_prefix("track.")?.strip_suffix(".jsonl")?; + valid_day(day).then_some(day) +} + +fn valid_day(day: &str) -> bool { + NaiveDate::parse_from_str(day, "%Y-%m-%d").is_ok() +} + +fn parse_time(timestamp: &str) -> Option> { + DateTime::parse_from_rfc3339(timestamp) + .ok() + .map(|timestamp| timestamp.with_timezone(&Utc)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event(id: &str, ts: &str) -> String { + serde_json::json!({"event_id": id, "ts": ts}).to_string() + "\n" + } + + #[test] + fn delayed_events_publish_under_their_utc_event_days() { + let mut index = EventPartitionIndex::empty("edge-m-codex"); + let raw = event("old", "2026-07-22T23:55:00Z") + &event("new", "2026-07-24T01:00:00+01:00"); + let groups = partition_lines(&mut index, raw.as_bytes(), "2026-07-24"); + assert_eq!( + groups.keys().cloned().collect::>(), + ["2026-07-22", "2026-07-24"] + ); + assert_eq!( + index.partitions["2026-07-22"].min_ts, + "2026-07-22T23:55:00Z" + ); + } + + #[test] + fn unknown_timestamps_fail_open_to_a_conservative_physical_day() { + let mut index = EventPartitionIndex::empty("edge-m-codex"); + let groups = partition_lines(&mut index, b"{\"event_id\":\"unknown\"}\n", "2026-07-24"); + assert!(groups.contains_key("2026-07-24")); + assert!(index.legacy_days.contains("2026-07-24")); + } + + #[test] + fn partition_ranges_compare_normalized_instants_not_offset_strings() { + let mut index = EventPartitionIndex::empty("edge-m-codex"); + let raw = event("late", "2026-07-22T12:30:00+02:00") + + &event("early", "2026-07-22T10:00:00Z"); + partition_lines(&mut index, raw.as_bytes(), "2026-07-22"); + + assert_eq!( + index.partitions["2026-07-22"].min_ts, + "2026-07-22T10:00:00Z" + ); + assert_eq!( + index.partitions["2026-07-22"].max_ts, + "2026-07-22T10:30:00Z" + ); + } + + #[test] + fn partition_ranges_order_whole_and_fractional_seconds_by_instant() { + let mut range = EventTimeRange::default(); + range.record("2026-07-22T10:00:00.500Z"); + range.record("2026-07-22T10:00:00Z"); + + assert_eq!(range.min_ts, "2026-07-22T10:00:00Z"); + assert_eq!(range.max_ts, "2026-07-22T10:00:00.500Z"); + } + + #[test] + fn object_watermarks_advance_per_day_and_never_regress() { + let mut index = EventPartitionIndex::empty("edge-m-codex"); + let day = "2026-07-22"; + let first = "events/edge-m-codex/chunks/track.2026-07-22/b.jsonl"; + index.record_object(day, first); + assert_eq!(index.object_cursors[day], first); + + index.record_object( + day, + "events/edge-m-codex/chunks/track.2026-07-22/a.jsonl", + ); + assert_eq!( + index.object_cursors[day], first, + "a late smaller key must not move the reader watermark backward" + ); + + let next_day = "2026-07-23"; + let next = "events/edge-m-codex/chunks/track.2026-07-23/a.jsonl"; + index.record_object(next_day, next); + assert_eq!(index.object_cursors[next_day], next); + } + + #[test] + fn bucket_freshness_uses_the_stream_registry_and_parsed_instants() { + let root = + std::env::temp_dir().join(format!("synty-event-freshness-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + for (stream, timestamp) in [ + ("edge-a-codex", "2026-07-22T10:00:00Z"), + ("edge-b-codex", "2026-07-22T10:00:00.500Z"), + ] { + bucket + .put(&format!("{}/{stream}", crate::sync::EVENT_STREAMS), b"") + .unwrap(); + bucket + .put( + &key(stream), + serde_json::to_vec(&serde_json::json!({ + "format": 1, + "stream": stream, + "partitions": { + "2026-07-22": {"min_ts": timestamp, "max_ts": timestamp} + } + })) + .unwrap() + .as_slice(), + ) + .unwrap(); + } + bucket + .put( + "event-partitions/ignored/track.2026-07-22.json", + b"not a stream index", + ) + .unwrap(); + + assert_eq!( + bucket_newest_event(&bucket).unwrap().as_deref(), + Some("2026-07-22T10:00:00.500Z") + ); + let _ = std::fs::remove_dir_all(&root); + } + + #[test] + fn object_ranges_prune_chunks_inside_a_broad_legacy_day() { + let mut index = EventObjectIndex::empty("edge-m-codex", "2026-07-22"); + record_object( + &mut index, + "events/edge-m-codex/chunks/track.2026-07-22/old.jsonl", + event("old", "2026-06-21T10:00:00Z").as_bytes(), + ); + record_object( + &mut index, + "events/edge-m-codex/chunks/track.2026-07-22/new.jsonl", + event("new", "2026-07-22T10:00:00Z").as_bytes(), + ); + + assert_eq!( + index.candidate_objects( + parse_time("2026-07-22T09:00:00Z").unwrap(), + parse_time("2026-07-22T11:00:00Z").unwrap(), + ), + BTreeSet::from([ + "events/edge-m-codex/chunks/track.2026-07-22/new.jsonl".to_string() + ]) + ); + } + + #[test] + fn malformed_object_coverage_stays_unindexed() { + let mut index = EventObjectIndex::empty("edge-m-codex", "2026-07-22"); + record_object( + &mut index, + "events/edge-m-codex/chunks/track.2026-07-22/unknown.jsonl", + b"{\"event_id\":\"unknown\"}\n", + ); + assert!(index.objects.is_empty()); + } + + #[test] + fn legacy_days_are_unconditionally_queried_until_their_range_is_complete() { + let mut index = EventPartitionIndex::empty("edge-m-codex"); + index.legacy_days.insert("2026-07-20".into()); + index.complete_legacy_day("2026-07-21", "2026-07-21T09:00:00Z", "2026-07-21T10:00:00Z"); + let since = parse_time("2026-07-22T00:00:00Z").unwrap(); + let until = parse_time("2026-07-23T00:00:00Z").unwrap(); + assert_eq!( + index.candidate_days(since, until), + BTreeSet::from(["2026-07-20".to_string()]) + ); + } + + #[test] + fn initialization_marks_existing_physical_days_as_legacy() { + let root = + std::env::temp_dir().join(format!("synty-event-partitions-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + bucket + .put( + "events/edge-m-codex/chunks/track.2026-07-21/one.jsonl", + b"{}\n", + ) + .unwrap(); + let index = load_or_initialize(&bucket, "edge-m-codex").unwrap(); + assert_eq!( + index.legacy_days, + BTreeSet::from(["2026-07-21".to_string()]) + ); + assert!(load(&bucket, "edge-m-codex").unwrap().is_some()); + let _ = std::fs::remove_dir_all(&root); + } +} diff --git a/src/main.rs b/src/main.rs index 2a3724b..af55650 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,7 @@ mod encode; mod eval; mod event; mod config; +mod event_partitions; mod fleet; mod github; mod generation; diff --git a/src/mcp.rs b/src/mcp.rs index e8c6252..bf57cba 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -9,7 +9,7 @@ use crate::{encode::Encoder, load_docs, readmodel, search, trace, units, view}; use anyhow::Result; use next_plaid::{MmapIndex, SearchParameters}; use serde_json::{json, Value}; -use std::io::{BufRead, Write}; +use std::{io::{BufRead, Write}, sync::{OnceLock, RwLock}}; pub(crate) const PROTOCOL_VERSION: &str = "2025-11-25"; pub(crate) const SUPPORTED_PROTOCOL_VERSIONS: &[&str] = @@ -422,12 +422,113 @@ impl Server { } } -/// Keep the selected published model fresh without holding the dispatcher. -/// Athena readers omit trace.json; neither mode mirrors the raw event lake. +#[derive(Clone, Debug, Default)] +pub(crate) struct BucketFreshness { + pub newest_raw_event: Option, + pub published_read_model: Option, + pub read_model_format: Option, + pub checked_at: Option, + pub error: Option, +} + +impl BucketFreshness { + #[cfg_attr(not(any(feature = "mcp-http", test)), allow(dead_code))] + pub(crate) fn remote_format_ready(&self) -> bool { + self.read_model_format + .is_some_and(|format| format >= readmodel::FORMAT) + } +} + +static BUCKET_FRESHNESS: OnceLock> = OnceLock::new(); + +pub(crate) fn bucket_freshness() -> BucketFreshness { + BUCKET_FRESHNESS + .get_or_init(|| RwLock::new(BucketFreshness::default())) + .read() + .map(|freshness| freshness.clone()) + .unwrap_or_default() +} + +fn refresh_bucket_freshness(bucket_uri: &str) -> Result { + let started = std::time::Instant::now(); + let result = (|| { + let bucket = crate::bucket::open(bucket_uri)?; + let current = bucket + .get("current.json")? + .map(|raw| serde_json::from_slice::(&raw)) + .transpose()?; + let mut freshness = BucketFreshness { + published_read_model: current.as_ref().and_then(|current| { + (!current.published_at.is_empty()).then(|| current.published_at.clone()) + }), + read_model_format: current.as_ref().map(|current| current.format), + checked_at: Some( + chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true), + ), + ..Default::default() + }; + match crate::event_partitions::bucket_newest_event(bucket.as_ref()) { + Ok(timestamp) => freshness.newest_raw_event = timestamp, + Err(error) => freshness.error = Some(error.to_string()), + } + Ok(freshness) + })(); + let (outcome, pointer_format, raw_events) = match &result { + Ok(freshness) => ( + if freshness.error.is_some() { + "partial" + } else { + "success" + }, + freshness.read_model_format.is_some(), + freshness.newest_raw_event.is_some(), + ), + Err(_) => ("error", false, false), + }; + crate::metrics::Run::new("bucket_freshness") + .set("outcome", outcome) + .set("elapsed_ms", started.elapsed().as_millis() as u64) + .set("pointer_format", pointer_format) + .set("raw_events", raw_events) + .emit(); + result +} + +fn record_bucket_freshness(freshness: BucketFreshness) { + if let Ok(mut current) = BUCKET_FRESHNESS + .get_or_init(|| RwLock::new(BucketFreshness::default())) + .write() + { + *current = freshness; + } +} + +/// Keep the selected published model and its remote freshness metadata current +/// without holding a dispatcher. Athena readers omit trace.json and inspect +/// only the small partition indexes; neither mode mirrors the raw event lake. pub(crate) fn start_bucket_refresh(bucket: Option, include_trace: bool) { let Some(bucket) = bucket else { return }; std::thread::spawn(move || loop { crate::sync::pull_read_model_for_mcp(&bucket, include_trace); + match refresh_bucket_freshness(&bucket) { + Ok(freshness) => { + if let Some(error) = freshness.error.as_deref() { + eprintln!("synty mcp: bucket freshness partially unavailable: {error}"); + } + record_bucket_freshness(freshness); + } + Err(error) => { + eprintln!("synty mcp: bucket freshness unavailable: {error:#}"); + record_bucket_freshness(BucketFreshness { + checked_at: Some( + chrono::Utc::now() + .to_rfc3339_opts(chrono::SecondsFormat::Secs, true), + ), + error: Some(error.to_string()), + ..Default::default() + }); + } + } std::thread::sleep(std::time::Duration::from_secs(30)); }); } @@ -752,4 +853,81 @@ mod tests { assert_eq!(resp["result"]["isError"], true); assert!(resp["result"]["content"][0]["text"].as_str().unwrap().contains("query")); } + + #[test] + fn bucket_freshness_reads_only_pointer_and_partition_metadata() { + use crate::bucket::Bucket; + + let root = + std::env::temp_dir().join(format!("synty-bucket-freshness-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + bucket + .put( + "current.json", + br#"{"build":"b","rev":1,"format":2,"writer":"0.2.5","published_at":"2026-07-22T12:00:00Z"}"#, + ) + .unwrap(); + bucket + .put("event-streams/edge-m-codex", b"") + .unwrap(); + bucket + .put( + "event-partitions/edge-m-codex.json", + br#"{"format":1,"stream":"edge-m-codex","partitions":{"2026-07-22":{"min_ts":"2026-07-22T10:00:00Z","max_ts":"2026-07-22T11:00:00Z"}}}"#, + ) + .unwrap(); + + let freshness = refresh_bucket_freshness(root.to_str().unwrap()).unwrap(); + + assert_eq!(freshness.read_model_format, Some(2)); + assert_eq!( + freshness.published_read_model.as_deref(), + Some("2026-07-22T12:00:00Z") + ); + assert_eq!( + freshness.newest_raw_event.as_deref(), + Some("2026-07-22T11:00:00Z") + ); + let _ = std::fs::remove_dir_all(&root); + } + + #[test] + fn malformed_raw_metadata_preserves_compatible_pointer_readiness() { + use crate::bucket::Bucket; + + let root = std::env::temp_dir().join(format!( + "synty-bucket-partial-freshness-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + bucket + .put( + "current.json", + br#"{"build":"b","rev":1,"format":2,"writer":"0.2.5","published_at":"2026-07-22T12:00:00Z"}"#, + ) + .unwrap(); + bucket + .put("event-streams/edge-m-codex", b"") + .unwrap(); + bucket + .put("event-partitions/edge-m-codex.json", b"{not-json") + .unwrap(); + + let freshness = refresh_bucket_freshness(root.to_str().unwrap()).unwrap(); + + assert_eq!(freshness.read_model_format, Some(readmodel::FORMAT)); + assert_eq!( + freshness.published_read_model.as_deref(), + Some("2026-07-22T12:00:00Z") + ); + assert!(freshness.newest_raw_event.is_none()); + assert!(freshness.error.is_some()); + assert!( + freshness.remote_format_ready(), + "a raw-index diagnostic must not make a compatible published model unready" + ); + let _ = std::fs::remove_dir_all(&root); + } } diff --git a/src/mcp_http.rs b/src/mcp_http.rs index 286d92d..759f02b 100644 --- a/src/mcp_http.rs +++ b/src/mcp_http.rs @@ -416,10 +416,13 @@ async fn handle( let mediated = crate::readmodel::mediated_ready(state.require_trace_projection); let dispatchers_ready = state.dispatcher.is_alive() && state.analysis_dispatcher.is_alive(); + let freshness = mcp::bucket_freshness(); + let remote_format_ready = freshness.remote_format_ready(); let ready = ready_state( state.require_read_model, read_model, mediated, + remote_format_ready, dispatchers_ready, ); let liveness = path == "/health"; @@ -428,6 +431,11 @@ async fn handle( "ready": ready, "read_model": read_model, "mediated_projections": mediated, + "bucket_newest_event": freshness.newest_raw_event, + "bucket_published_read_model": freshness.published_read_model, + "bucket_read_model_format": freshness.read_model_format, + "bucket_checked_at": freshness.checked_at, + "bucket_freshness_error": health_freshness_error(freshness.error.as_deref()), "dispatchers": dispatchers_ready, "trace_backend": if state.require_trace_projection { "projection" } else { "athena" }, "name": "synty", @@ -561,14 +569,20 @@ fn health_code(ready: bool) -> u16 { if ready { 200 } else { 503 } } +#[cfg(any(feature = "mcp-http", test))] +fn health_freshness_error(error: Option<&str>) -> Option<&'static str> { + error.map(|_| "bucket metadata unavailable") +} + #[cfg(any(feature = "mcp-http", test))] fn ready_state( require_read_model: bool, read_model: bool, mediated: bool, + remote_format: bool, dispatchers: bool, ) -> bool { - dispatchers && (!require_read_model || (read_model && mediated)) + dispatchers && (!require_read_model || (read_model && mediated && remote_format)) } #[cfg(any(feature = "mcp-http", test))] @@ -734,9 +748,19 @@ mod tests { assert!(!protocol_version_supported(Some("2099-01-01"))); assert_eq!(health_code(true), 200); assert_eq!(health_code(false), 503); - assert!(ready_state(true, true, true, true)); - assert!(!ready_state(true, true, true, false)); - assert!(!ready_state(true, true, false, true)); + assert_eq!(health_freshness_error(None), None); + assert_eq!( + health_freshness_error(Some( + "AccessDenied: bucket super-secret-bucket in account 123" + )), + Some("bucket metadata unavailable"), + "unauthenticated health responses expose no provider diagnostics" + ); + assert!(ready_state(true, true, true, true, true)); + assert!(!ready_state(true, true, true, true, false)); + assert!(!ready_state(true, true, false, true, true)); + assert!(!ready_state(true, true, true, false, true)); + assert!(ready_state(false, false, false, false, true)); } #[test] diff --git a/src/readmodel.rs b/src/readmodel.rs index f74bb4b..75fa4c6 100644 --- a/src/readmodel.rs +++ b/src/readmodel.rs @@ -31,6 +31,10 @@ pub struct Current { /// The synty version that wrote this pointer — for debugging mixed fleets. #[serde(default)] pub writer: String, + /// Wall-clock publication time for remote freshness reporting. Old + /// pointers deserialize with an empty value and remain readable. + #[serde(default)] + pub published_at: String, } fn format_default() -> u32 { @@ -77,20 +81,31 @@ pub fn current() -> Option { /// Atomically repoint readers at `build`/`rev`. The build directory must be /// complete before this is called — the pointer move IS the publish. pub fn repoint(build: &str, rev: u64) -> Result<()> { + repoint_at(Path::new(POINTER), build, rev) +} + +fn repoint_at(pointer: &Path, build: &str, rev: u64) -> Result<()> { let cur = Current { build: build.into(), rev, format: FORMAT, writer: env!("CARGO_PKG_VERSION").into(), + published_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true), }; - repoint_current(&cur) + repoint_current_at(pointer, &cur) } /// Preserve the exact pointer advertised by a remote writer. In particular, /// pulling a legacy build must not relabel it as the current layout format. pub fn repoint_current(cur: &Current) -> Result<()> { - std::fs::create_dir_all(ROOT)?; - crate::write_atomic(POINTER, serde_json::to_string(cur)?.as_bytes()) + repoint_current_at(Path::new(POINTER), cur) +} + +fn repoint_current_at(pointer: &Path, cur: &Current) -> Result<()> { + if let Some(parent) = pointer.parent() { + std::fs::create_dir_all(parent)?; + } + crate::write_atomic(&pointer.to_string_lossy(), serde_json::to_string(cur)?.as_bytes()) } /// Reader conveniences: resolve through the pointer, with the working-corpus @@ -221,7 +236,7 @@ mod tests { // docs/clusters paths through it (rev selects the clusters file). #[test] fn pointer_resolves_versioned_paths() { - let c = Current { build: "abc123".into(), rev: 2, format: FORMAT, writer: String::new() }; + let c = Current { build: "abc123".into(), rev: 2, format: FORMAT, writer: String::new(), published_at: String::new() }; assert_eq!(c.dir(), Path::new("index/builds/abc123")); assert!(c.docs().ends_with("index/builds/abc123/docs.jsonl")); assert!(c.clusters().ends_with("index/builds/abc123/unit_clusters.2.json")); @@ -242,6 +257,26 @@ mod tests { assert_eq!(old, newer, "identity is (build, rev) only"); } + #[test] + fn publishing_a_pointer_persists_an_rfc3339_publication_time() { + let root = + std::env::temp_dir().join(format!("synty-repoint-time-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let pointer = root.join("index/current.json"); + + repoint_at(&pointer, "published-build", 3).unwrap(); + + let current: Current = + serde_json::from_slice(&std::fs::read(&pointer).unwrap()).unwrap(); + assert!(!current.published_at.is_empty()); + assert!( + chrono::DateTime::parse_from_rfc3339(¤t.published_at).is_ok(), + "published_at must be an RFC3339 instant: {}", + current.published_at + ); + let _ = std::fs::remove_dir_all(&root); + } + #[test] fn build_id_is_stable_and_distinguishing() { assert_eq!(build_id(&[1, 2, 3], &[7, 8]), build_id(&[1, 2, 3], &[7, 8])); @@ -266,6 +301,7 @@ mod tests { rev: 0, format: FORMAT, writer: String::new(), + published_at: String::new(), }; assert!(mediated_ready_from(Some(current.clone()), false)); diff --git a/src/sync.rs b/src/sync.rs index 6cd6823..94d8bc5 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -8,7 +8,7 @@ // pointing at a complete build. The per-doc embeddings (large, build-side, // already in the content-addressed store) are not published. -use crate::{bucket, readmodel}; +use crate::{bucket, event_partitions, readmodel}; use anyhow::{Result, anyhow}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -17,7 +17,7 @@ use std::path::Path; const POINTER_KEY: &str = "current.json"; const EVENTS: &str = "events"; -const EVENT_STREAMS: &str = "event-streams"; +pub(crate) const EVENT_STREAMS: &str = "event-streams"; const EVENT_CHUNK_BYTES: usize = 1 << 20; /// Filename → blob hash for one (build, rev). @@ -262,6 +262,7 @@ fn push_events_scoped( let allowed_by_stream = state.allowed_sessions.entry(bucket_uri.to_string()).or_default(); let (mut n, mut bytes_up) = (0, 0u64); let mut changed = profile_changed || repo_policy_changed; + let mut partition_indexes = BTreeMap::new(); let mut files: Vec = walkdir::WalkDir::new(local_dir) .into_iter() .filter_map(|e| e.ok()) @@ -333,22 +334,54 @@ fn push_events_scoped( filter_events_for_sessions(&filtered, allowed) }; let redacted = redact_event_lines(&repo_filtered, profile); - for (part, chunk) in event_chunks(&redacted, EVENT_CHUNK_BYTES) - .into_iter() - .enumerate() - { - let hash = Sha256::digest(&chunk); - let short_hash = hash[..8] - .iter() - .map(|b| format!("{b:02x}")) - .collect::(); - let stem = file.strip_suffix(".jsonl").unwrap_or(file); - let key = format!( - "{EVENTS}/{stream}/chunks/{stem}/{start:016x}-{end:016x}-{part:04}-{short_hash}.jsonl" + if !partition_indexes.contains_key(stream) { + partition_indexes.insert( + stream.to_string(), + event_partitions::load_or_initialize(b.as_ref(), stream)?, ); - if b.put_if_absent(&key, &chunk)? { - bytes_up += chunk.len() as u64; - n += 1; + } + let partition_index = partition_indexes + .get_mut(stream) + .expect("partition index initialized"); + let fallback_day = event_partitions::day_from_track_file(file) + .map(str::to_string) + .unwrap_or_else(|| chrono::Utc::now().format("%Y-%m-%d").to_string()); + let partitioned = + event_partitions::partition_lines(partition_index, &redacted, &fallback_day); + // Publish coverage before the chunks. A reader may briefly query an + // empty projected day, but it can never miss a newly-addressable day. + event_partitions::save(b.as_ref(), partition_index)?; + for (day, body) in partitioned { + let mut object_index = + event_partitions::load_objects_or_empty(b.as_ref(), stream, &day)?; + let mut objects = Vec::new(); + for (part, chunk) in event_chunks(&body, EVENT_CHUNK_BYTES) + .into_iter() + .enumerate() + { + let hash = Sha256::digest(&chunk); + let short_hash = hash[..8] + .iter() + .map(|b| format!("{b:02x}")) + .collect::(); + let stem = format!("track.{day}"); + let key = format!( + "{EVENTS}/{stream}/chunks/{stem}/{start:016x}-{end:016x}-{part:04}-{short_hash}.jsonl" + ); + event_partitions::record_object(&mut object_index, &key, &chunk); + partition_index.record_object(&day, &key); + objects.push((key, chunk)); + } + // Like the day range, exact object coverage is visible first. A + // referenced-but-not-yet-present path scans nothing; the inverse + // ordering could briefly make a new object invisible. + event_partitions::save_objects(b.as_ref(), &object_index)?; + event_partitions::save(b.as_ref(), partition_index)?; + for (key, chunk) in objects { + if b.put_if_absent(&key, &chunk)? { + bytes_up += chunk.len() as u64; + n += 1; + } } } start = end; @@ -588,6 +621,16 @@ fn event_chunks(raw: &[u8], target: usize) -> Vec> { struct DownloadState { #[serde(default)] buckets: BTreeMap>, + /// Event-day writers may publish a newly observed older day after a newer + /// one. Per-day cursors preserve delta reads without assuming stream keys + /// are globally append-sorted. + #[serde(default)] + partitions: BTreeMap>, + /// Streams whose pre-watermark history was scanned once by this reader. + /// New writers advertise exact per-day object cursors, so future pulls can + /// skip the global compatibility LIST until an old writer is observed. + #[serde(default)] + compatibility_scanned: BTreeMap>, } /// Download all devices' event files from the bucket into `local_dir`. Writers @@ -611,6 +654,11 @@ fn pull_events_from( .and_then(|s| serde_json::from_str(&s).ok()) .unwrap_or_default(); let cursors = state.buckets.entry(bucket_uri.to_string()).or_default(); + let partition_cursors = state.partitions.entry(bucket_uri.to_string()).or_default(); + let compatibility_scanned = state + .compatibility_scanned + .entry(bucket_uri.to_string()) + .or_default(); let (mut n, mut bytes_down) = (0, 0u64); let markers = b.list(EVENT_STREAMS)?; let mut cursor_changed = false; @@ -630,18 +678,70 @@ fn pull_events_from( if stream.is_empty() || stream.contains('/') { continue; } - let chunk_prefix = format!("{EVENTS}/{stream}/chunks/"); - let mut cursor = cursors.get(stream).cloned().unwrap_or_default(); - if !cursor.is_empty() && !event_dest(local_dir, &cursor)?.is_file() { - cursor.clear(); // local corpus was pruned; reconstruct it - } - for key in b.list_after(&chunk_prefix, &cursor)? { - let (fetched, bytes) = pull_event_key(b, &key, local_dir)?; - n += usize::from(fetched); - bytes_down += bytes; - cursor = key; - cursors.insert(stream.to_string(), cursor.clone()); - cursor_changed = true; + if let Some(index) = event_partitions::load(b, stream)? { + // A new-layout writer publishes an exact high-water key per + // event day. One global scan seeds a reader with any history + // that predates those watermarks; after that, only changed + // days need a LIST. If an older writer later rewrites the + // metadata without the layout marker, resume the compatibility + // cursor until the new writer takes over again. + if index.writer_layout == 0 || !compatibility_scanned.contains(stream) { + let chunk_prefix = format!("{EVENTS}/{stream}/chunks/"); + let mut cursor = cursors.get(stream).cloned().unwrap_or_default(); + if !cursor.is_empty() && !event_dest(local_dir, &cursor)?.is_file() { + cursor.clear(); + } + for key in b.list_after(&chunk_prefix, &cursor)? { + let (fetched, bytes) = pull_event_key(b, &key, local_dir)?; + n += usize::from(fetched); + bytes_down += bytes; + cursor = key.clone(); + cursors.insert(stream.to_string(), cursor.clone()); + record_partition_cursor(partition_cursors, stream, &key); + cursor_changed = true; + } + if index.writer_layout > 0 + && compatibility_scanned.insert(stream.to_string()) + { + cursor_changed = true; + } + } + for (day, remote_cursor) in &index.object_cursors { + let partition = format!("{stream}\0{day}"); + let chunk_prefix = format!("{EVENTS}/{stream}/chunks/track.{day}/"); + let mut cursor = partition_cursors + .get(&partition) + .cloned() + .unwrap_or_default(); + if !cursor.is_empty() && !event_dest(local_dir, &cursor)?.is_file() { + cursor.clear(); + } + if !cursor.is_empty() && cursor >= *remote_cursor { + continue; + } + for key in b.list_after(&chunk_prefix, &cursor)? { + let (fetched, bytes) = pull_event_key(b, &key, local_dir)?; + n += usize::from(fetched); + bytes_down += bytes; + cursor = key; + partition_cursors.insert(partition.clone(), cursor.clone()); + cursor_changed = true; + } + } + } else { + let chunk_prefix = format!("{EVENTS}/{stream}/chunks/"); + let mut cursor = cursors.get(stream).cloned().unwrap_or_default(); + if !cursor.is_empty() && !event_dest(local_dir, &cursor)?.is_file() { + cursor.clear(); // local corpus was pruned; reconstruct it + } + for key in b.list_after(&chunk_prefix, &cursor)? { + let (fetched, bytes) = pull_event_key(b, &key, local_dir)?; + n += usize::from(fetched); + bytes_down += bytes; + cursor = key; + cursors.insert(stream.to_string(), cursor.clone()); + cursor_changed = true; + } } // Old mutable daily files sit beside `chunks/`, so this bounded // compatibility list never walks the immutable chunk history. @@ -662,6 +762,22 @@ fn pull_events_from( Ok(n) } +fn record_partition_cursor( + partition_cursors: &mut BTreeMap, + stream: &str, + key: &str, +) { + let Some(day) = event_partitions::day_from_event_key(key) else { + return; + }; + let cursor = partition_cursors + .entry(format!("{stream}\0{day}")) + .or_default(); + if key > cursor.as_str() { + *cursor = key.to_string(); + } +} + fn event_dest(local_dir: &str, key: &str) -> Result { let rel = key.strip_prefix(&format!("{EVENTS}/")).unwrap_or(key); if !Path::new(rel) @@ -964,12 +1080,14 @@ mod tests { rev: 0, format: readmodel::FORMAT, writer: "new".into(), + published_at: "2026-07-22T10:00:00Z".into(), }; let remote = readmodel::Current { build: "remote-v1".into(), rev: 9, format: 1, writer: "old".into(), + published_at: String::new(), }; assert!(local_format_is_newer(Some(&local), &remote)); assert!(!local_format_is_newer(None, &remote)); @@ -1100,6 +1218,15 @@ mod tests { !first[0].ends_with("track.2026-07-21.jsonl"), "growing daily file is never the cloud object" ); + let object_index = event_partitions::load_objects( + &b, + "edge-m-codex", + "2026-07-21", + ) + .unwrap() + .unwrap(); + assert_eq!(object_index.objects.len(), 1); + assert!(object_index.objects.contains_key(&first[0])); let first_bytes = b.get(&first[0]).unwrap().unwrap(); use std::io::Write; @@ -1133,6 +1260,15 @@ mod tests { 2, "append creates one new object" ); + assert_eq!( + event_partitions::load_objects(&b, "edge-m-codex", "2026-07-21") + .unwrap() + .unwrap() + .objects + .len(), + 2, + "object coverage advances with immutable chunks" + ); assert_eq!( b.get(&first[0]).unwrap().unwrap(), first_bytes, @@ -1303,8 +1439,8 @@ mod tests { Some(cutoff), ) .unwrap(), - 1, - "first eligible delta carries staged metadata" + 2, + "first eligible delta carries staged metadata in its own event day" ); let b = crate::bucket::LocalFs::new(&bucket); let body = b @@ -1369,7 +1505,12 @@ mod tests { ).unwrap(), 0); let calls = store.calls.lock().unwrap().clone(); assert!(!calls.iter().any(|c| c == "list:events"), "historical event namespace was relisted: {calls:?}"); - assert!(calls.iter().any(|c| c.starts_with("after:events/edge-m-codex/chunks/:events/edge-m-codex/chunks/")), "missing persisted stream cursor: {calls:?}"); + assert!( + !calls + .iter() + .any(|call| call.starts_with("after:events/edge-m-codex/chunks/")), + "unchanged historical days must not issue LIST calls: {calls:?}" + ); use std::io::Write; std::fs::OpenOptions::new().append(true).open(&file).unwrap() @@ -1381,12 +1522,133 @@ mod tests { upload_state.to_str().unwrap(), None, ).unwrap(), 1); + store.calls.lock().unwrap().clear(); assert_eq!(pull_events_from( &store, bucket.to_str().unwrap(), reader.to_str().unwrap(), &download_state, ).unwrap(), 1); + let calls = store.calls.lock().unwrap().clone(); + assert_eq!( + calls + .iter() + .filter(|call| call.starts_with("after:events/edge-m-codex/chunks/")) + .count(), + 1, + "only the day whose object watermark advanced should be listed: {calls:?}" + ); + assert!( + calls.iter().any(|call| call.starts_with( + "after:events/edge-m-codex/chunks/track.2026-07-21/" + )), + "the changed day must continue from its persisted cursor: {calls:?}" + ); + let _ = std::fs::remove_dir_all(&root); + } + + #[test] + fn a_late_older_event_day_is_visible_after_a_newer_partition_cursor() { + let root = + std::env::temp_dir().join(format!("synty-event-late-day-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let source = root.join("source/edge-m-codex"); + let bucket = root.join("bucket"); + let reader = root.join("reader"); + std::fs::create_dir_all(&source).unwrap(); + std::fs::write( + source.join("track.2026-07-23.jsonl"), + event("newer", "s", "user_prompt", "2026-07-23T10:00:00Z", "newer"), + ) + .unwrap(); + let upload_state = root.join("uploads.json"); + push_events_with( + bucket.to_str().unwrap(), + root.join("source").to_str().unwrap(), + upload_state.to_str().unwrap(), + None, + ) + .unwrap(); + let download_state = reader.join(".downloads.json"); + let store = crate::bucket::LocalFs::new(&bucket); + assert_eq!( + pull_events_from( + &store, + bucket.to_str().unwrap(), + reader.to_str().unwrap(), + &download_state, + ) + .unwrap(), + 1 + ); + + std::fs::write( + source.join("track.2026-07-21.jsonl"), + event( + "late", + "s", + "assistant_message", + "2026-07-21T10:00:00Z", + "late", + ), + ) + .unwrap(); + push_events_with( + bucket.to_str().unwrap(), + root.join("source").to_str().unwrap(), + upload_state.to_str().unwrap(), + None, + ) + .unwrap(); + assert_eq!( + pull_events_from( + &store, + bucket.to_str().unwrap(), + reader.to_str().unwrap(), + &download_state, + ) + .unwrap(), + 1, + "per-day cursors must discover a newly published older day" + ); + let body = crate::units::jsonl_files(&reader.join("edge-m-codex")) + .into_iter() + .filter_map(|path| std::fs::read_to_string(path).ok()) + .collect::(); + assert!(body.contains("\"event_id\":\"late\""), "{body}"); + let mut old_writer_index = + event_partitions::load(&store, "edge-m-codex").unwrap().unwrap(); + old_writer_index.writer_layout = 0; + store + .put( + &event_partitions::key("edge-m-codex"), + &serde_json::to_vec(&old_writer_index).unwrap(), + ) + .unwrap(); + store + .put( + "events/edge-m-codex/chunks/track.2026-07-24/legacy-writer.jsonl", + event( + "legacy", + "s", + "assistant_message", + "2026-07-24T10:00:00Z", + "legacy", + ) + .as_bytes(), + ) + .unwrap(); + assert_eq!( + pull_events_from( + &store, + bucket.to_str().unwrap(), + reader.to_str().unwrap(), + &download_state, + ) + .unwrap(), + 1, + "the compatibility cursor must discover an old writer's new day" + ); let _ = std::fs::remove_dir_all(&root); } diff --git a/src/trace_athena.rs b/src/trace_athena.rs index ad11edd..ee90d02 100644 --- a/src/trace_athena.rs +++ b/src/trace_athena.rs @@ -2,12 +2,13 @@ // lake by injected stream partitions and a bounded time window; the existing // trace fold then reconstructs turns, spans, and jobs from only those rows. -use crate::{bucket, metrics, policy::ReadScope, trace}; +use crate::{bucket, event_partitions, metrics, policy::ReadScope, trace}; use anyhow::{Context, Result, bail}; use aws_config::timeout::TimeoutConfig; use aws_sdk_athena::types::{QueryExecutionContext, QueryExecutionState}; use chrono::{DateTime, Duration, NaiveDate, Utc}; use std::collections::BTreeSet; +use std::fmt; use std::time::{Duration as StdDuration, Instant}; const DEFAULT_LIST_HOURS: i64 = 1; @@ -16,10 +17,36 @@ const MAX_LOOKBACK_HOURS: i64 = 24 * 7; const MAX_EVENTS: usize = 50_000; const MAX_RESULT_BYTES: usize = 64 * 1024 * 1024; const MAX_SESSIONS: usize = 200; -const QUERY_TIMEOUT: StdDuration = StdDuration::from_secs(50); +const MAX_OBJECT_PATHS: usize = 1_000; +const MAX_PATH_PREDICATE_BYTES: usize = 200_000; +const REQUEST_QUERY_TIMEOUT: StdDuration = StdDuration::from_secs(45); const AWS_CONNECT_TIMEOUT: StdDuration = StdDuration::from_secs(3); const AWS_OPERATION_TIMEOUT: StdDuration = StdDuration::from_secs(10); +#[derive(Debug)] +enum TraceQueryError { + Timeout(String), + Limit(String), +} + +impl fmt::Display for TraceQueryError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Timeout(message) | Self::Limit(message) => formatter.write_str(message), + } + } +} + +impl std::error::Error for TraceQueryError {} + +fn timeout_error(message: impl Into) -> anyhow::Error { + TraceQueryError::Timeout(message.into()).into() +} + +fn limit_error(message: impl Into) -> anyhow::Error { + TraceQueryError::Limit(message.into()).into() +} + #[derive(Clone, Debug)] pub(crate) struct Config { pub bucket: String, @@ -55,8 +82,13 @@ struct QueryRows { lines: Vec, } +struct ObjectSelection { + days: Vec, + paths: Vec, +} + trait EventQuery: Send { - fn run(&mut self, sql: &str) -> Result; + fn run(&mut self, sql: &str, deadline: Instant) -> Result; } struct AwsEventQuery { @@ -91,139 +123,172 @@ impl AwsEventQuery { database: config.database.clone(), }) } + + fn stop_execution(&self, id: &str) { + let _ = self.runtime.block_on( + self.client + .stop_query_execution() + .query_execution_id(id) + .send(), + ); + } } impl EventQuery for AwsEventQuery { - fn run(&mut self, sql: &str) -> Result { - anyhow::ensure!( - sql.trim_start().to_ascii_uppercase().starts_with("SELECT "), - "Athena trace only permits SELECT statements" - ); + fn run(&mut self, sql: &str, deadline: Instant) -> Result { let started = Instant::now(); - let execution = self - .runtime - .block_on( - self.client - .start_query_execution() - .work_group(&self.workgroup) - .query_execution_context( - QueryExecutionContext::builder() - .database(&self.database) - .build(), - ) - .query_string(sql) - .send(), - ) - .context("start Athena trace query")?; - let id = execution - .query_execution_id() - .context("Athena returned no query execution id")? - .to_string(); - - let scanned_bytes = loop { - if started.elapsed() >= QUERY_TIMEOUT { - let _ = self.runtime.block_on( - self.client - .stop_query_execution() - .query_execution_id(&id) - .send(), - ); - bail!( - "Athena trace query timed out after {} seconds", - QUERY_TIMEOUT.as_secs() - ); + let mut row_count = 0usize; + let mut result_bytes = 0usize; + let mut scanned_bytes = 0i64; + let result = (|| { + anyhow::ensure!( + sql.trim_start().to_ascii_uppercase().starts_with("SELECT "), + "Athena trace only permits SELECT statements" + ); + if Instant::now() >= deadline { + return Err(timeout_error( + "Athena trace request timed out before query execution", + )); } let execution = self .runtime .block_on( self.client - .get_query_execution() - .query_execution_id(&id) + .start_query_execution() + .work_group(&self.workgroup) + .query_execution_context( + QueryExecutionContext::builder() + .database(&self.database) + .build(), + ) + .query_string(sql) .send(), ) - .context("poll Athena trace query")?; - let query = execution - .query_execution() - .context("Athena returned no query execution")?; - let status = query.status().context("Athena returned no query status")?; - match status.state() { - Some(QueryExecutionState::Succeeded) => { - break query - .statistics() - .and_then(|statistics| statistics.data_scanned_in_bytes()) - .unwrap_or(0); + .context("start Athena trace query")?; + let id = execution + .query_execution_id() + .context("Athena returned no query execution id")? + .to_string(); + + scanned_bytes = loop { + if Instant::now() >= deadline { + self.stop_execution(&id); + return Err(timeout_error( + "Athena trace request timed out after its shared query budget", + )); } - Some(QueryExecutionState::Failed | QueryExecutionState::Cancelled) => { - bail!( - "Athena trace query {}: {}", - status - .state() - .map(|state| state.as_str()) - .unwrap_or("failed"), - status.state_change_reason().unwrap_or("no reason returned") - ); + let poll = (|| { + let execution = self + .runtime + .block_on( + self.client + .get_query_execution() + .query_execution_id(&id) + .send(), + ) + .context("poll Athena trace query")?; + let query = execution + .query_execution() + .context("Athena returned no query execution")?; + let status = query.status().context("Athena returned no query status")?; + match status.state() { + Some(QueryExecutionState::Succeeded) => Ok(Some( + query + .statistics() + .and_then(|statistics| statistics.data_scanned_in_bytes()) + .unwrap_or(0), + )), + Some(QueryExecutionState::Failed | QueryExecutionState::Cancelled) => { + bail!( + "Athena trace query {}: {}", + status + .state() + .map(|state| state.as_str()) + .unwrap_or("failed"), + status.state_change_reason().unwrap_or("no reason returned") + ) + } + _ => Ok(None), + } + })(); + match poll { + Ok(Some(bytes)) => break bytes, + Ok(None) => std::thread::sleep(StdDuration::from_millis(250)), + Err(error) => { + self.stop_execution(&id); + return Err(error); + } } - _ => std::thread::sleep(StdDuration::from_millis(250)), - } - }; + }; - let mut lines = Vec::new(); - let mut bytes = 0usize; - let mut next_token = None; - let mut first_page = true; - loop { - anyhow::ensure!( - started.elapsed() < QUERY_TIMEOUT, - "Athena trace result retrieval timed out after {} seconds", - QUERY_TIMEOUT.as_secs() - ); - let mut request = self - .client - .get_query_results() - .query_execution_id(&id) - .max_results(1000); - if let Some(token) = next_token.as_deref() { - request = request.next_token(token); - } - let page = self - .runtime - .block_on(request.send()) - .context("read Athena trace query results")?; - if let Some(result_set) = page.result_set() { - for (index, row) in result_set.rows().iter().enumerate() { - if first_page && index == 0 { - continue; + let mut lines = Vec::new(); + let mut next_token = None; + let mut first_page = true; + loop { + if Instant::now() >= deadline { + return Err(timeout_error( + "Athena trace result retrieval timed out after its shared query budget", + )); + } + let mut request = self + .client + .get_query_results() + .query_execution_id(&id) + .max_results(1000); + if let Some(token) = next_token.as_deref() { + request = request.next_token(token); + } + let page = self + .runtime + .block_on(request.send()) + .context("read Athena trace query results")?; + if let Some(result_set) = page.result_set() { + for (index, row) in result_set.rows().iter().enumerate() { + if first_page && index == 0 { + continue; + } + let Some(line) = + row.data().first().and_then(|datum| datum.var_char_value()) + else { + continue; + }; + result_bytes = result_bytes.saturating_add(line.len()); + if result_bytes > MAX_RESULT_BYTES { + return Err(limit_error(format!( + "Athena trace selection exceeds {} MiB; narrow the time, machine, source, or operation filter", + MAX_RESULT_BYTES / 1024 / 1024 + ))); + } + lines.push(line.to_string()); + row_count = lines.len(); + if lines.len() > MAX_EVENTS { + return Err(limit_error(format!( + "Athena trace selection exceeds {MAX_EVENTS} events; narrow the time, machine, source, or operation filter" + ))); + } } - let Some(line) = row.data().first().and_then(|datum| datum.var_char_value()) - else { - continue; - }; - bytes = bytes.saturating_add(line.len()); - anyhow::ensure!( - bytes <= MAX_RESULT_BYTES, - "Athena trace selection exceeds {} MiB; narrow the time, machine, source, or operation filter", - MAX_RESULT_BYTES / 1024 / 1024 - ); - lines.push(line.to_string()); - anyhow::ensure!( - lines.len() <= MAX_EVENTS, - "Athena trace selection exceeds {MAX_EVENTS} events; narrow the time, machine, source, or operation filter" - ); + } + first_page = false; + next_token = page.next_token().map(str::to_string); + if next_token.is_none() { + break; } } - first_page = false; - next_token = page.next_token().map(str::to_string); - if next_token.is_none() { - break; - } - } + Ok(QueryRows { lines }) + })(); + let outcome = result + .as_ref() + .err() + .map(query_outcome) + .unwrap_or("success"); metrics::Run::new("athena_trace") - .set("rows", lines.len()) - .set("result_bytes", bytes) + .set("outcome", outcome) + .set("rows", row_count) + .set("result_bytes", result_bytes) .set("scanned_bytes", scanned_bytes) .set("elapsed_ms", started.elapsed().as_millis() as u64) .emit(); - Ok(QueryRows { lines }) + result } } @@ -233,6 +298,9 @@ pub(crate) struct Backend { // Tests inject a fixed registry; production re-lists the bounded registry // for every tool call so newly enrolled streams appear without a restart. streams: Option>, + // Tests may inject physical partitions without opening a real bucket. + #[cfg(test)] + days: Option>, cached: Option, } @@ -243,6 +311,8 @@ impl Backend { config, query, streams: None, + #[cfg(test)] + days: None, cached: None, }) } @@ -263,6 +333,7 @@ impl Backend { limit: usize, scope: &ReadScope, ) -> Result { + let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT; let window = Window::parse(since, None, DEFAULT_LIST_HOURS)?; let store = self.load_store( window, @@ -273,6 +344,7 @@ impl Backend { &[], operation.is_some(), scope, + deadline, )?; let out = trace::list_store_text( &store, @@ -305,8 +377,10 @@ impl Backend { { return Ok(out); } + let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT; let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?; - let store = self.load_store(window, None, None, None, None, &[id], true, scope)?; + let store = + self.load_store(window, None, None, None, None, &[id], true, scope, deadline)?; let out = trace::show_store_text(&store, id, before, after, false, Some(scope))?; self.cached = Some(store); Ok(out) @@ -323,6 +397,7 @@ impl Backend { limit: usize, scope: &ReadScope, ) -> Result { + let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT; let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?; let store = self.load_store( window, @@ -333,6 +408,7 @@ impl Backend { &[], false, scope, + deadline, )?; let out = trace::search_store_text( &store, @@ -354,8 +430,19 @@ impl Backend { { return Ok(out); } + let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT; let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?; - let store = self.load_store(window, None, None, None, None, &[left, right], true, scope)?; + let store = self.load_store( + window, + None, + None, + None, + None, + &[left, right], + true, + scope, + deadline, + )?; let out = trace::compare_store_text(&store, left, right, false, Some(scope))?; self.cached = Some(store); Ok(out) @@ -372,22 +459,24 @@ impl Backend { ids: &[&str], expand_matching_sessions: bool, scope: &ReadScope, + deadline: Instant, ) -> Result { - let streams = self.selected_streams(machine)?; + let streams = self.selected_streams(machine, source, scope)?; let predicate = Predicate { source: source.map(str::to_string), scope_sources: scope.sources.clone(), needle: needle.map(str::to_string), kind_contains: kind.map(str::to_string), - ids: ids.iter().map(|id| (*id).to_string()).collect(), + ids: ids.iter().map(|id| query_id(id).to_string()).collect(), ..Default::default() }; - let first = self.select(&streams, window, &predicate)?; + let first = self.select(&streams, window, &predicate, deadline)?; let sessions = event_sessions(&first.lines); - anyhow::ensure!( - sessions.len() <= MAX_SESSIONS, - "Athena trace selection spans more than {MAX_SESSIONS} sessions; narrow the time, machine, source, or operation filter" - ); + if sessions.len() > MAX_SESSIONS { + return Err(limit_error(format!( + "Athena trace selection spans more than {MAX_SESSIONS} sessions; narrow the time, machine, source, or operation filter" + ))); + } let context_window = Window { since: std::cmp::max( window.since - Duration::days(1), @@ -406,6 +495,7 @@ impl Backend { scope_sources: scope.sources.clone(), ..Default::default() }, + deadline, )? .lines } else { @@ -423,6 +513,7 @@ impl Backend { dedupe_session_kinds: true, ..Default::default() }, + deadline, )? .lines; contexts.append(&mut lines); @@ -436,12 +527,37 @@ impl Backend { streams: &[String], window: Window, predicate: &Predicate, + deadline: Instant, ) -> Result { - let sql = select_sql(&self.config, streams, window, predicate, MAX_EVENTS + 1)?; - self.query.run(&sql) + let selection = self.selected_objects(streams, window)?; + if selection.paths.is_empty() { + metrics::Run::new("athena_trace") + .set("outcome", "empty") + .set("rows", 0) + .set("result_bytes", 0) + .set("scanned_bytes", 0) + .set("elapsed_ms", 0) + .emit(); + return Ok(QueryRows { lines: Vec::new() }); + } + let sql = select_sql( + &self.config, + streams, + &selection.days, + &selection.paths, + window, + predicate, + MAX_EVENTS + 1, + )?; + self.query.run(&sql, deadline) } - fn selected_streams(&mut self, machine: Option<&str>) -> Result> { + fn selected_streams( + &mut self, + machine: Option<&str>, + source: Option<&str>, + scope: &ReadScope, + ) -> Result> { let mut streams = if let Some(streams) = &self.streams { streams.clone() } else { @@ -458,16 +574,117 @@ impl Backend { ); streams }; - if let Some(machine) = machine { - let machine = machine.to_ascii_lowercase(); - streams.retain(|stream| stream.to_ascii_lowercase().contains(&machine)); - } + let machine = machine.map(str::to_ascii_lowercase); + let requested_source = source.map(normalized_stream_source); + let scope_sources = scope + .sources + .iter() + .map(|source| normalized_stream_source(source)) + .collect::>(); + streams.retain(|stream| { + let Some((stream_machine, stream_source)) = crate::identity::stream_parts(stream) + else { + return machine.is_none() && requested_source.is_none() && scope_sources.is_empty(); + }; + machine + .as_deref() + .is_none_or(|wanted| stream_machine.eq_ignore_ascii_case(wanted)) + && requested_source + .as_deref() + .is_none_or(|wanted| stream_source.eq_ignore_ascii_case(wanted)) + && (scope_sources.is_empty() + || scope_sources.contains(&normalized_stream_source(stream_source))) + }); anyhow::ensure!( !streams.is_empty(), - "no event streams match the machine filter" + "no event streams match the machine, source, and read-scope filters" ); Ok(streams) } + + fn selected_objects(&self, streams: &[String], window: Window) -> Result { + #[cfg(test)] + if let Some(days) = &self.days { + let stream = streams.first().context("Athena trace needs a test stream")?; + return Ok(ObjectSelection { + days: days.clone(), + paths: days + .iter() + .map(|day| { + format!( + "{}/events/{stream}/chunks/track.{day}/test.jsonl", + self.config.bucket.trim_end_matches('/') + ) + }) + .collect(), + }); + } + let bucket = bucket::open(&self.config.bucket)?; + let mut selected = BTreeSet::new(); + for stream in streams { + if let Some(index) = event_partitions::load(bucket.as_ref(), stream)? { + for day in index.candidate_days(window.since, window.until) { + let physical_day = bucket + .list(&format!("events/{stream}/chunks/track.{day}/"))? + .into_iter() + .collect::>(); + if let Some(objects) = + event_partitions::load_objects(bucket.as_ref(), stream, &day)? + { + let indexed_objects = + objects.objects.keys().cloned().collect::>(); + selected.extend( + objects.candidate_objects(window.since, window.until), + ); + selected.extend( + physical_day + .difference(&indexed_objects) + .cloned(), + ); + } else { + selected.extend(physical_day); + } + } + } else { + let days = window_days(window); + selected.extend( + bucket + .list(&format!("events/{stream}/chunks/"))? + .into_iter() + .filter(|key| { + event_partitions::day_from_event_key(key) + .is_some_and(|day| days.contains(day)) + }), + ); + } + } + if selected.len() > MAX_OBJECT_PATHS { + return Err(limit_error(format!( + "Athena trace selection needs {} raw objects, exceeding the {MAX_OBJECT_PATHS}-object query limit; narrow the machine, source, or time filter, or backfill object-range metadata", + selected.len() + ))); + } + let path_bytes = selected + .iter() + .map(|key| key.len() + self.config.bucket.len() + 4) + .sum::(); + if path_bytes > MAX_PATH_PREDICATE_BYTES { + return Err(limit_error(format!( + "Athena trace object paths need {path_bytes} SQL bytes, exceeding the {MAX_PATH_PREDICATE_BYTES}-byte path-predicate limit; narrow the machine, source, or time filter" + ))); + } + let days = selected + .iter() + .filter_map(|key| event_partitions::day_from_event_key(key).map(str::to_string)) + .collect::>() + .into_iter() + .collect(); + let paths = selected + .into_iter() + .map(|key| format!("{}/{}", self.config.bucket.trim_end_matches('/'), key)) + .collect(); + Ok(ObjectSelection { days, paths }) + } } #[derive(Default)] @@ -499,10 +716,11 @@ impl Window { None => until - Duration::hours(default_hours), }; anyhow::ensure!(since < until, "trace since must be before until"); - anyhow::ensure!( - until - since <= Duration::hours(MAX_LOOKBACK_HOURS), - "Athena trace windows are limited to {MAX_LOOKBACK_HOURS} hours" - ); + if until - since > Duration::hours(MAX_LOOKBACK_HOURS) { + return Err(limit_error(format!( + "Athena trace windows are limited to {MAX_LOOKBACK_HOURS} hours" + ))); + } Ok(Self { since, until }) } } @@ -522,6 +740,8 @@ fn parse_time(value: &str) -> Result> { fn select_sql( config: &Config, streams: &[String], + days: &[String], + paths: &[String], window: Window, predicate: &Predicate, limit: usize, @@ -530,18 +750,27 @@ fn select_sql( !streams.is_empty(), "Athena trace needs at least one stream" ); + anyhow::ensure!(!days.is_empty(), "Athena trace needs at least one day"); + anyhow::ensure!(!paths.is_empty(), "Athena trace needs at least one object path"); let stream_values = streams .iter() .map(|stream| sql_string(stream)) .collect::>() .join(", "); + let day_values = days + .iter() + .map(|day| sql_string(day)) + .collect::>() + .join(", "); + let path_values = paths + .iter() + .map(|path| sql_string(path)) + .collect::>() + .join(", "); let mut clauses = vec![ format!("stream IN ({stream_values})"), - format!( - "day BETWEEN {} AND {}", - sql_string(&window.since.format("%Y-%m-%d").to_string()), - sql_string(&window.until.format("%Y-%m-%d").to_string()) - ), + format!("day IN ({day_values})"), + format!("\"$path\" IN ({path_values})"), format!( "from_iso8601_timestamp(json_extract_scalar(line, '$.ts')) >= from_iso8601_timestamp({})", sql_string(&window.since.to_rfc3339()) @@ -656,6 +885,49 @@ fn sql_string(value: &str) -> String { format!("'{}'", value.replace('\'', "''")) } +fn query_id(id: &str) -> &str { + id.strip_prefix("job:").unwrap_or(id) +} + +fn normalized_stream_source(source: &str) -> String { + match source.to_ascii_lowercase().as_str() { + "codex" | "codex_cli" | "codex-cli" => "codex".into(), + "claude" | "claudecode" | "claude_code" | "claude-code" => "claudecode".into(), + "cowork" => "cowork".into(), + "harness" => "harness".into(), + "devin" => "devin".into(), + other => other.to_string(), + } +} + +fn window_days(window: Window) -> BTreeSet { + let mut days = BTreeSet::new(); + let mut day = window.since.date_naive(); + loop { + let start = day + .and_hms_opt(0, 0, 0) + .expect("midnight exists") + .and_utc(); + if start >= window.until { + break; + } + days.insert(day.format("%Y-%m-%d").to_string()); + let Some(next) = day.succ_opt() else { + break; + }; + day = next; + } + days +} + +fn query_outcome(error: &anyhow::Error) -> &'static str { + match error.downcast_ref::() { + Some(TraceQueryError::Timeout(_)) => "timeout", + Some(TraceQueryError::Limit(_)) => "limit", + None => "error", + } +} + fn validate_name(label: &str, value: &str, lowercase_only: bool) -> Result<()> { let valid = !value.is_empty() && value.len() <= 128 @@ -683,7 +955,7 @@ mod tests { } impl EventQuery for FakeQuery { - fn run(&mut self, sql: &str) -> Result { + fn run(&mut self, sql: &str, _deadline: Instant) -> Result { self.calls.lock().unwrap().push(sql.to_string()); Ok(QueryRows { lines: self.lines.clone(), @@ -712,6 +984,11 @@ mod tests { let sql = select_sql( &config, &["edge-m-codex".into()], + &["2026-07-20".into(), "2026-07-22".into()], + &[ + "s3://bucket/events/edge-m-codex/chunks/track.2026-07-20/a.jsonl".into(), + "s3://bucket/events/edge-m-codex/chunks/track.2026-07-22/b.jsonl".into(), + ], Window { since: parse_time("2026-07-22T10:00:00Z").unwrap(), until: parse_time("2026-07-22T11:00:00Z").unwrap(), @@ -722,7 +999,8 @@ mod tests { .unwrap(); assert!(sql.starts_with("SELECT line")); assert!(sql.contains("stream IN ('edge-m-codex')")); - assert!(sql.contains("day BETWEEN '2026-07-22' AND '2026-07-22'")); + assert!(sql.contains("day IN ('2026-07-20', '2026-07-22')")); + assert!(sql.contains("\"$path\" IN ('s3://bucket/events/edge-m-codex/")); for mutating in ["INSERT", "UPDATE", "DELETE", "CREATE", "UNLOAD", "CTAS"] { assert!(!sql.to_ascii_uppercase().contains(mutating), "{mutating}"); } @@ -765,6 +1043,7 @@ mod tests { calls: Arc::clone(&calls), }), streams: Some(vec!["edge-m-codex".into()]), + days: Some(vec!["2026-07-22".into()]), cached: None, }; let out = backend @@ -785,6 +1064,10 @@ mod tests { .unwrap(); assert!(out.contains("exec_command")); assert!(out.contains("call-1")); + assert!( + backend.cached.is_none(), + "a list slice must not satisfy a later show or compare lookup" + ); assert!( calls .lock() @@ -825,6 +1108,7 @@ mod tests { calls: Arc::clone(&calls), }), streams: Some(vec!["edge-m-codex".into()]), + days: Some(vec!["2026-07-22".into()]), cached: None, }; @@ -885,6 +1169,7 @@ mod tests { calls: Arc::clone(&calls), }), streams: Some(vec!["edge-m-codex".into()]), + days: Some(vec!["2026-07-22".into()]), cached: None, }; @@ -897,6 +1182,273 @@ mod tests { assert!(!calls[1].contains("synty_context_rank")); } + #[test] + fn listed_job_ids_query_the_native_span_id() { + let lines = vec![ + event( + "call-1", + "2026-07-22T10:00:01Z", + "tool_call", + json!({"name":"exec_command","call_id":"c1","arguments":"{\"cmd\":\"cargo test\"}"}), + ), + event( + "result-1", + "2026-07-22T10:00:03Z", + "tool_result", + json!({"call_id":"c1","output":"Process running with session ID 42"}), + ), + ]; + let calls = Arc::new(Mutex::new(Vec::new())); + let config = Config::new( + "s3://bucket".into(), + "wg".into(), + "synty".into(), + "raw_events".into(), + ) + .unwrap(); + let mut backend = Backend { + config, + query: Box::new(FakeQuery { + lines, + calls: Arc::clone(&calls), + }), + streams: Some(vec!["edge-m-codex".into()]), + days: Some(vec!["2026-07-22".into()]), + cached: None, + }; + + let out = backend + .show("job:call-1", 3, 5, &ReadScope::default()) + .unwrap(); + + assert!(out.contains("cargo test"), "{out}"); + let calls = calls.lock().unwrap(); + assert!(calls[0].contains("'call-1'")); + assert!(!calls[0].contains("'job:call-1'")); + } + + #[test] + fn stream_pruning_uses_exact_machine_and_canonical_source_suffixes() { + let calls = Arc::new(Mutex::new(Vec::new())); + let config = Config::new( + "s3://bucket".into(), + "wg".into(), + "synty".into(), + "raw_events".into(), + ) + .unwrap(); + let mut backend = Backend { + config, + query: Box::new(FakeQuery { + lines: Vec::new(), + calls, + }), + streams: Some(vec![ + "edge-dev-box-codex".into(), + "edge-dev-box-claudecode".into(), + "edge-dev-box-2-codex".into(), + ]), + days: Some(vec!["2026-07-22".into()]), + cached: None, + }; + let scope = ReadScope { + sources: vec!["codex_cli".into()], + ..Default::default() + }; + + assert_eq!( + backend + .selected_streams(Some("dev-box"), Some("codex_cli"), &scope) + .unwrap(), + vec!["edge-dev-box-codex"] + ); + assert!( + backend + .selected_streams(None, Some("unknown-agent"), &ReadScope::default()) + .is_err(), + "an explicit unknown source must fail closed" + ); + assert!( + backend + .selected_streams( + None, + None, + &ReadScope { + sources: vec!["unknown-agent".into()], + ..Default::default() + }, + ) + .is_err(), + "an all-unknown source scope must fail closed" + ); + } + + #[test] + fn partition_ranges_select_delayed_and_unknown_physical_days() { + use crate::bucket::Bucket; + + let root = + std::env::temp_dir().join(format!("synty-athena-partitions-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + for day in ["2026-07-19", "2026-07-20", "2026-07-24"] { + bucket + .put( + &format!("events/edge-m-codex/chunks/track.{day}/one.jsonl"), + b"{}\n", + ) + .unwrap(); + } + bucket + .put( + "events/edge-m-codex/chunks/track.2026-07-20/old.jsonl", + b"{}\n", + ) + .unwrap(); + bucket + .put( + "event-partitions/edge-m-codex.json", + serde_json::to_string(&json!({ + "format": 1, + "stream": "edge-m-codex", + "legacy_days": ["2026-07-19", "2026-07-24"], + "partitions": { + "2026-07-20": { + "min_ts": "2026-07-22T10:00:00Z", + "max_ts": "2026-07-22T11:00:00Z" + } + }, + "updated_at": "2026-07-24T12:00:00Z" + })) + .unwrap() + .as_bytes(), + ) + .unwrap(); + bucket + .put( + "event-partitions/edge-m-codex/track.2026-07-20.json", + serde_json::to_string(&json!({ + "format": 1, + "stream": "edge-m-codex", + "day": "2026-07-20", + "objects": { + "events/edge-m-codex/chunks/track.2026-07-20/one.jsonl": { + "min_ts": "2026-07-22T10:00:00Z", + "max_ts": "2026-07-22T11:00:00Z" + }, + "events/edge-m-codex/chunks/track.2026-07-20/old.jsonl": { + "min_ts": "2026-06-21T10:00:00Z", + "max_ts": "2026-06-21T11:00:00Z" + } + } + })) + .unwrap() + .as_bytes(), + ) + .unwrap(); + let backend = Backend { + config: Config { + bucket: root.to_string_lossy().into_owned(), + workgroup: "wg".into(), + database: "synty".into(), + table: "raw_events".into(), + }, + query: Box::new(FakeQuery { + lines: Vec::new(), + calls: Arc::new(Mutex::new(Vec::new())), + }), + streams: Some(vec!["edge-m-codex".into()]), + days: None, + cached: None, + }; + let selection = backend + .selected_objects( + &["edge-m-codex".into()], + Window { + since: parse_time("2026-07-22T10:30:00Z").unwrap(), + until: parse_time("2026-07-22T10:45:00Z").unwrap(), + }, + ) + .unwrap(); + + assert_eq!( + selection.days, + ["2026-07-19", "2026-07-20", "2026-07-24"] + ); + assert_eq!(selection.paths.len(), 3); + assert!(!selection.paths.iter().any(|path| path.ends_with("/old.jsonl"))); + let _ = std::fs::remove_dir_all(&root); + } + + #[test] + fn query_metrics_classify_timeouts_and_limits() { + let timeout = Err::<(), _>(timeout_error("request timed out")) + .context("Athena request") + .unwrap_err(); + assert_eq!( + query_outcome(&timeout), + "timeout" + ); + assert_eq!( + query_outcome(&limit_error("selection exceeds 50000 events")), + "limit" + ); + assert_eq!( + query_outcome(&anyhow::anyhow!("AWS rejected the request")), + "error" + ); + } + + #[test] + fn legacy_streams_limit_object_paths_to_the_requested_physical_days() { + use crate::bucket::Bucket; + + let root = std::env::temp_dir().join(format!( + "synty-athena-legacy-window-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + for day in ["2026-07-20", "2026-07-22", "2026-07-23"] { + bucket + .put( + &format!("events/edge-m-codex/chunks/track.{day}/one.jsonl"), + b"{}\n", + ) + .unwrap(); + } + let backend = Backend { + config: Config { + bucket: root.to_string_lossy().into_owned(), + workgroup: "wg".into(), + database: "synty".into(), + table: "raw_events".into(), + }, + query: Box::new(FakeQuery { + lines: Vec::new(), + calls: Arc::new(Mutex::new(Vec::new())), + }), + streams: Some(vec!["edge-m-codex".into()]), + days: None, + cached: None, + }; + + let selection = backend + .selected_objects( + &["edge-m-codex".into()], + Window { + since: parse_time("2026-07-22T10:00:00Z").unwrap(), + until: parse_time("2026-07-23T00:00:00Z").unwrap(), + }, + ) + .unwrap(); + + assert_eq!(selection.days, ["2026-07-22"]); + assert_eq!(selection.paths.len(), 1); + assert!(selection.paths[0].contains("track.2026-07-22")); + let _ = std::fs::remove_dir_all(&root); + } + #[test] fn windows_fail_closed_before_scanning_more_than_seven_days() { let error = Window::parse( diff --git a/src/track.rs b/src/track.rs index 2bdde1f..f6155c3 100644 --- a/src/track.rs +++ b/src/track.rs @@ -14,7 +14,7 @@ use crate::event::{kind, Event, Sequencer}; use crate::tail::{drive, ms_to_rfc3339, EmitCtx, Source}; use anyhow::{anyhow, bail, Result}; use serde_json::json; -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::io::Write; use std::path::{Path, PathBuf}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -422,23 +422,28 @@ impl Stream { if events.is_empty() { return Ok(()); } - // Per-day local file in the stream dir. Sync reads only appended bytes - // and publishes them as immutable chunks on its own cadence. + // Partition by the envelope's UTC event day, not the wall clock when a + // delayed source line happened to be tailed. Sync preserves the same + // event-day partition in the bucket. std::fs::create_dir_all(&self.out)?; - let path = self.out.join(format!( - "track.{}.jsonl", - day_stamp(unix_ms(SystemTime::now())) - )); - let mut f = std::fs::OpenOptions::new() - .create(true) - .append(true) - .open(&path)?; - let mut body = String::new(); + let fallback = day_stamp(unix_ms(SystemTime::now())); + let mut by_day: BTreeMap = BTreeMap::new(); for e in events { + let day = event_time_ms(e) + .map(day_stamp) + .unwrap_or_else(|| fallback.clone()); + let body = by_day.entry(day).or_default(); body.push_str(&serde_json::to_string(e).map_err(|e| anyhow!("encode: {e}"))?); body.push('\n'); } - f.write_all(body.as_bytes())?; + for (day, body) in by_day { + let path = self.out.join(format!("track.{day}.jsonl")); + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path)?; + file.write_all(body.as_bytes())?; + } Ok(()) } } @@ -1145,6 +1150,50 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[test] + fn delayed_events_are_written_to_their_utc_event_day() { + let dir = + std::env::temp_dir().join(format!("synty-track-event-day-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let stream = Stream { + src: Box::new(ClaudeCode), + roots: vec![], + name: "edge-t-claudecode".into(), + out: dir.clone(), + seq: Sequencer::new(), + started: HashSet::new(), + files: HashMap::new(), + open: HashMap::new(), + n_sessions: 0, + n_skipped: 0, + actor: "tester".into(), + campaign: String::new(), + campaign_role: String::new(), + }; + let make = |id: &str, ts: &str| Event { + v: crate::event::ENVELOPE_V, + event_id: id.into(), + stream: "edge-t-claudecode".into(), + seq: 0, + ts: ts.into(), + source: crate::event::source::CLAUDE_CODE.into(), + session_id: "session".into(), + kind: kind::USER_PROMPT.into(), + payload: json!({"text": id}), + rollup_dim: String::new(), + }; + stream + .append(&[ + make("old", "2026-07-21T23:59:00Z"), + make("new", "2026-07-23T00:01:00Z"), + ]) + .unwrap(); + assert!(dir.join("track.2026-07-21.jsonl").is_file()); + assert!(dir.join("track.2026-07-23.jsonl").is_file()); + assert!(!dir.join("track.2026-07-22.jsonl").exists()); + let _ = std::fs::remove_dir_all(&dir); + } + // Concatenate a stream dir's per-day files — the tracker now writes // track..jsonl, so a test reads the dir, not a fixed filename. fn read_stream(dir: &std::path::Path) -> String { diff --git a/src/tui.rs b/src/tui.rs index ce67c8f..513b716 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -193,6 +193,11 @@ impl Bundle { autostart: false, bucket: None, upgrade: None, + bucket_newest_event: None, + bucket_published_read_model: None, + bucket_read_model_format: None, + bucket_checked_at: None, + bucket_freshness_error: None, stale: false, fleet: Default::default(), }); @@ -2054,7 +2059,7 @@ mod tests { sessions: vec![session], work, topics, - status: crate::view::Status { docs: 2, github: 1, sessions: 1, by_kind: vec![("user_prompt".into(), 1), ("pull_request".into(), 1)], by_repo: vec![crate::view::Tally { name: "sie".into(), docs: 1, github: 0, sessions: 1, tok_out: 18_900, tools: 8 }], by_user: vec![crate::view::Tally { name: "alice".into(), docs: 1, github: 1, sessions: 0, tok_out: 0, tools: 0 }], by_tool: vec![crate::view::ToolTally { name: "Bash".into(), agent: "claude".into(), calls: 5, errs: 1, chars: 81_200 }, crate::view::ToolTally { name: "Edit".into(), agent: "claude".into(), calls: 3, errs: 0, chars: 0 }], by_model: vec![units::ModelUsage { model: "claude-fable-5".into(), tok_in: 4_200, tok_out: 18_900, cache_read: 310_000, cache_create: 12_000, turns: 7 }], newest_ts: "2026-05-31".into(), last_indexed: None, last_tracked: None, autostart: false, bucket: None, upgrade: None, stale: false, fleet: crate::fleet::Roster { machines: vec![crate::fleet::Machine { machine: "mac-3939".into(), sources: vec!["claude".into(), "codex".into()], actors: vec!["svonava".into()], last_ts: "2026-05-31T09:00:00Z".into(), version: "0.1.0".into(), events: 41, quiet: false }, crate::fleet::Machine { machine: "ci-runner-7".into(), sources: vec!["claude".into()], actors: vec![], last_ts: "2026-05-12T09:00:00Z".into(), version: String::new(), events: 0, quiet: true }], actors_tracked: vec!["svonava".into()], gh_active: 2, untracked: vec![crate::fleet::UntrackedAuthor { login: "bob".into(), agent: Some("claude".into()) }], install_rate_pct: 50, quiet_days: 7 } }, + status: crate::view::Status { docs: 2, github: 1, sessions: 1, by_kind: vec![("user_prompt".into(), 1), ("pull_request".into(), 1)], by_repo: vec![crate::view::Tally { name: "sie".into(), docs: 1, github: 0, sessions: 1, tok_out: 18_900, tools: 8 }], by_user: vec![crate::view::Tally { name: "alice".into(), docs: 1, github: 1, sessions: 0, tok_out: 0, tools: 0 }], by_tool: vec![crate::view::ToolTally { name: "Bash".into(), agent: "claude".into(), calls: 5, errs: 1, chars: 81_200 }, crate::view::ToolTally { name: "Edit".into(), agent: "claude".into(), calls: 3, errs: 0, chars: 0 }], by_model: vec![units::ModelUsage { model: "claude-fable-5".into(), tok_in: 4_200, tok_out: 18_900, cache_read: 310_000, cache_create: 12_000, turns: 7 }], newest_ts: "2026-05-31".into(), last_indexed: None, last_tracked: None, autostart: false, bucket: None, upgrade: None, bucket_newest_event: None, bucket_published_read_model: None, bucket_read_model_format: None, bucket_checked_at: None, bucket_freshness_error: None, stale: false, fleet: crate::fleet::Roster { machines: vec![crate::fleet::Machine { machine: "mac-3939".into(), sources: vec!["claude".into(), "codex".into()], actors: vec!["svonava".into()], last_ts: "2026-05-31T09:00:00Z".into(), version: "0.1.0".into(), events: 41, quiet: false }, crate::fleet::Machine { machine: "ci-runner-7".into(), sources: vec!["claude".into()], actors: vec![], last_ts: "2026-05-12T09:00:00Z".into(), version: String::new(), events: 0, quiet: true }], actors_tracked: vec!["svonava".into()], gh_active: 2, untracked: vec![crate::fleet::UntrackedAuthor { login: "bob".into(), agent: Some("claude".into()) }], install_rate_pct: 50, quiet_days: 7 } }, view: View::Topics, sel: 0, nav: Nav::default(), diff --git a/src/view.rs b/src/view.rs index fb56ef7..b172395 100644 --- a/src/view.rs +++ b/src/view.rs @@ -62,6 +62,13 @@ pub struct Status { /// A newer synty version published to the bucket, or None (current, or no /// bucket). Drives the passive upgrade nag; never blocks anything. pub upgrade: Option, + /// Remote raw-event and published-model freshness, sampled from small + /// bucket metadata objects by MCP readers without downloading the lake. + pub bucket_newest_event: Option, + pub bucket_published_read_model: Option, + pub bucket_read_model_format: Option, + pub bucket_checked_at: Option, + pub bucket_freshness_error: Option, /// Tracked events are newer than the index — answers may miss recent work. pub stale: bool, /// Per-machine liveness and the actor↔GitHub-author join (M8 coverage). @@ -127,6 +134,7 @@ pub fn status() -> Result { // A newer binary published to GitHub Releases (cached, token-gated, best- // effort). Independent of the bucket — it's about synty itself, not the data. let bucket = crate::config::load().bucket; + let bucket_freshness = crate::mcp::bucket_freshness(); let upgrade = crate::release::available(); let fleet = crate::units::analysis_roster() .unwrap_or_else(|| crate::fleet::roster(&docs, std::path::Path::new(crate::units::LOCAL_DIR))); @@ -145,6 +153,11 @@ pub fn status() -> Result { autostart: crate::track::autostart_enabled(), bucket, upgrade, + bucket_newest_event: bucket_freshness.newest_raw_event, + bucket_published_read_model: bucket_freshness.published_read_model, + bucket_read_model_format: bucket_freshness.read_model_format, + bucket_checked_at: bucket_freshness.checked_at, + bucket_freshness_error: bucket_freshness.error, stale: stale_note().is_some(), fleet, }) @@ -829,6 +842,24 @@ pub fn status_md(s: &Status) -> String { o.push_str(&line); o.push('\n'); } + if s.bucket.is_some() { + o.push_str(&format!( + "bucket raw: {} · published model: {} · format: {} · checked: {}{}\n", + s.bucket_newest_event.as_deref().unwrap_or("unknown"), + s.bucket_published_read_model + .as_deref() + .unwrap_or("unknown"), + s.bucket_read_model_format + .map(|format| format.to_string()) + .as_deref() + .unwrap_or("unknown"), + s.bucket_checked_at.as_deref().unwrap_or("pending"), + s.bucket_freshness_error + .as_deref() + .map(|error| format!(" · error: {error}")) + .unwrap_or_default(), + )); + } o.push_str(&format!( "{} docs · {} GitHub · {} sessions · autostart {}\nnewest: {}\nlast indexed: {} · last tracked: {}\n\nkinds: ", s.docs, @@ -965,40 +996,48 @@ pub fn topics_json(topics: &[TopicUnits]) -> String { } pub fn status_json(s: &Status) -> String { - envelope("status", serde_json::json!({ - "docs": s.docs, - "github": s.github, - "sessions": s.sessions, - "by_kind": s.by_kind, - "by_repo": s.by_repo.iter().map(|t| serde_json::json!({"name": t.name, "docs": t.docs, "github": t.github, "sessions": t.sessions})).collect::>(), - "by_user": s.by_user.iter().map(|t| serde_json::json!({"name": t.name, "docs": t.docs, "github": t.github, "sessions": t.sessions})).collect::>(), - "newest_ts": s.newest_ts, - "last_indexed": s.last_indexed, - "last_tracked": s.last_tracked, - "autostart": s.autostart, - "bucket": s.bucket, - "upgrade": s.upgrade, - "stale": s.stale, - "fleet": { - "machines": s.fleet.machines.iter().map(|m| serde_json::json!({ - "machine": m.machine, - "sources": m.sources, - "actors": m.actors, - "last_ts": m.last_ts, - "tracker_version": m.version, - "events": m.events, - "quiet": m.quiet, - })).collect::>(), - "actors_tracked": s.fleet.actors_tracked, - "gh_active": s.fleet.gh_active, - "untracked": s.fleet.untracked.iter().map(|u| serde_json::json!({ - "login": u.login, "agent": u.agent, - })).collect::>(), - "untracked_attributed": s.fleet.untracked_attributed(), - "install_rate_pct": s.fleet.install_rate_pct, - "quiet_days": s.fleet.quiet_days, - }, - })) + envelope( + "status", + serde_json::json!({ + "docs": s.docs, + "github": s.github, + "sessions": s.sessions, + "by_kind": s.by_kind, + "by_repo": s.by_repo.iter().map(|t| serde_json::json!({"name": t.name, "docs": t.docs, "github": t.github, "sessions": t.sessions})).collect::>(), + "by_user": s.by_user.iter().map(|t| serde_json::json!({"name": t.name, "docs": t.docs, "github": t.github, "sessions": t.sessions})).collect::>(), + "newest_ts": s.newest_ts, + "last_indexed": s.last_indexed, + "last_tracked": s.last_tracked, + "autostart": s.autostart, + "bucket": s.bucket, + "upgrade": s.upgrade, + "bucket_newest_event": s.bucket_newest_event, + "bucket_published_read_model": s.bucket_published_read_model, + "bucket_read_model_format": s.bucket_read_model_format, + "bucket_checked_at": s.bucket_checked_at, + "bucket_freshness_error": s.bucket_freshness_error, + "stale": s.stale, + "fleet": { + "machines": s.fleet.machines.iter().map(|m| serde_json::json!({ + "machine": m.machine, + "sources": m.sources, + "actors": m.actors, + "last_ts": m.last_ts, + "tracker_version": m.version, + "events": m.events, + "quiet": m.quiet, + })).collect::>(), + "actors_tracked": s.fleet.actors_tracked, + "gh_active": s.fleet.gh_active, + "untracked": s.fleet.untracked.iter().map(|u| serde_json::json!({ + "login": u.login, "agent": u.agent, + })).collect::>(), + "untracked_attributed": s.fleet.untracked_attributed(), + "install_rate_pct": s.fleet.install_rate_pct, + "quiet_days": s.fleet.quiet_days, + }, + }), + ) } // ── helpers ─────────────────────────────────────────────────────────────── @@ -1661,17 +1700,52 @@ mod tests { last_indexed: None, last_tracked: None, autostart: false, - bucket: None, + bucket: Some("s3://team-synty".into()), upgrade: None, + bucket_newest_event: Some("2026-07-22T11:00:00Z".into()), + bucket_published_read_model: Some("2026-07-22T12:00:00Z".into()), + bucket_read_model_format: Some(2), + bucket_checked_at: Some("2026-07-22T12:01:00Z".into()), + bucket_freshness_error: Some("raw index unavailable".into()), stale: false, fleet: Default::default(), }; - let s: serde_json::Value = serde_json::from_str(&status_json(&s)).unwrap(); - assert_eq!((s["v"].as_i64(), s["kind"].as_str()), (Some(1), Some("status"))); - assert_eq!(s["data"]["docs"], 3, "the data keeps the prior object shape"); - assert!(s["data"]["fleet"]["machines"].is_array()); - // Activation + the upgrade nag are exposed to agents (null here). - let data = s["data"].as_object().unwrap(); + let markdown = status_md(&s); + for value in [ + "2026-07-22T11:00:00Z", + "2026-07-22T12:00:00Z", + "format: 2", + "2026-07-22T12:01:00Z", + "raw index unavailable", + ] { + assert!(markdown.contains(value), "missing {value} from:\n{markdown}"); + } + let json: serde_json::Value = serde_json::from_str(&status_json(&s)).unwrap(); + assert_eq!( + (json["v"].as_i64(), json["kind"].as_str()), + (Some(1), Some("status")) + ); + assert_eq!(json["data"]["docs"], 3, "the data keeps the prior object shape"); + assert!(json["data"]["fleet"]["machines"].is_array()); + assert_eq!( + json["data"]["bucket_newest_event"], + "2026-07-22T11:00:00Z" + ); + assert_eq!( + json["data"]["bucket_published_read_model"], + "2026-07-22T12:00:00Z" + ); + assert_eq!(json["data"]["bucket_read_model_format"], 2); + assert_eq!( + json["data"]["bucket_checked_at"], + "2026-07-22T12:01:00Z" + ); + assert_eq!( + json["data"]["bucket_freshness_error"], + "raw index unavailable" + ); + // Activation + the upgrade nag remain explicit fields for agents. + let data = json["data"].as_object().unwrap(); assert!(data.contains_key("bucket"), "activation in json"); assert!(data.contains_key("upgrade"), "upgrade availability in json"); }