MCP: bound Athena trace ID lookups

README.md

@@ -186,6 +186,10 @@ New uploads use each event's UTC day, while
 day and `event-partitions/<stream>/track.<day>.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.
+Exact `trace_show` and `trace_compare` lookups recover event time from Synty's
+ULID ids. If an exact id or resolved-session lookup still spans more object
+paths than the request guard permits, Athena falls back to the bounded
+stream/day partitions while retaining the exact id/session predicate.
 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

src/event.rs

@@ -106,6 +106,29 @@ pub fn deterministic_ulid(ts_ms: u64, key: &str) -> String {
     ulid_string(ts_ms, &entropy)
 }
 
+/// Recover the 48-bit millisecond timestamp from a canonical ULID. Trace
+/// lookups use this to prune the raw event lake before applying an exact id
+/// predicate; foreign/non-ULID ids deliberately return `None`.
+pub fn ulid_timestamp_ms(id: &str) -> Option<u64> {
+    if id.len() != 26 {
+        return None;
+    }
+    let mut timestamp = 0u64;
+    for byte in id.bytes().take(10) {
+        let value = match byte.to_ascii_uppercase() {
+            b'0'..=b'9' => byte - b'0',
+            b'A'..=b'H' => byte.to_ascii_uppercase() - b'A' + 10,
+            b'J'..=b'K' => byte.to_ascii_uppercase() - b'J' + 18,
+            b'M'..=b'N' => byte.to_ascii_uppercase() - b'M' + 20,
+            b'P'..=b'T' => byte.to_ascii_uppercase() - b'P' + 22,
+            b'V'..=b'Z' => byte.to_ascii_uppercase() - b'V' + 27,
+            _ => return None,
+        };
+        timestamp = (timestamp << 5) | u64::from(value);
+    }
+    (timestamp <= 0x0000_ffff_ffff_ffff).then_some(timestamp)
+}
+
 /// Encode (48-bit time, 80-bit entropy) into the canonical 26-char Crockford
 /// base32 ULID string (the oklog/ulid byte packing).
 fn ulid_string(ts_ms: u64, entropy: &[u8; 10]) -> String {
@@ -192,6 +215,16 @@ mod tests {
         assert!(early[..10] < late[..10]);
     }
 
+    #[test]
+    fn ulid_timestamp_roundtrips_and_rejects_foreign_ids() {
+        let timestamp = 1_700_000_000_123;
+        let id = deterministic_ulid(timestamp, "session");
+        assert_eq!(ulid_timestamp_ms(&id), Some(timestamp));
+        assert_eq!(ulid_timestamp_ms(&id.to_ascii_lowercase()), Some(timestamp));
+        assert_eq!(ulid_timestamp_ms("job:call-1"), None);
+        assert_eq!(ulid_timestamp_ms("Z0000000000000000000000000"), None);
+    }
+
     // The envelope round-trips and matches the wire field names ingest reads.
     #[test]
     fn event_json_uses_canonical_field_names() {

src/trace_athena.rs

@@ -13,6 +13,7 @@ use std::time::{Duration as StdDuration, Instant};
 
 const DEFAULT_LIST_HOURS: i64 = 1;
 const DEFAULT_LOOKUP_HOURS: i64 = 24 * 7;
+const ID_LOOKUP_MINUTES: i64 = 5;
 const MAX_LOOKBACK_HOURS: i64 = 24 * 7;
 const MAX_EVENTS: usize = 50_000;
 const MAX_RESULT_BYTES: usize = 64 * 1024 * 1024;
@@ -378,7 +379,8 @@ impl Backend {
             return Ok(out);
         }
         let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT;
-        let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?;
+        let window = id_lookup_window(id)
+            .unwrap_or(Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?);
         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))?;
@@ -431,7 +433,8 @@ impl Backend {
             return Ok(out);
         }
         let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT;
-        let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?;
+        let window = ids_lookup_window(&[left, right])
+            .unwrap_or(Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?);
         let store = self.load_store(
             window,
             None,
@@ -472,6 +475,16 @@ impl Backend {
         };
         let first = self.select(&streams, window, &predicate, deadline)?;
         let sessions = event_sessions(&first.lines);
+        let matched_streams = event_streams(&first.lines);
+        let context_streams = if matched_streams.is_empty() {
+            streams.clone()
+        } else {
+            streams
+                .iter()
+                .filter(|stream| matched_streams.contains(*stream))
+                .cloned()
+                .collect()
+        };
         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"
@@ -488,7 +501,7 @@ impl Backend {
             && !sessions.is_empty();
         let mut lines = if expands_sessions {
             self.select(
-                &streams,
+                &context_streams,
                 context_window,
                 &Predicate {
                     sessions: sessions.iter().cloned().collect(),
@@ -504,7 +517,7 @@ impl Backend {
         if !sessions.is_empty() && !expands_sessions {
             let mut contexts = self
                 .select(
-                    &streams,
+                    &context_streams,
                     context_window,
                     &Predicate {
                         sessions: sessions.into_iter().collect(),
@@ -529,8 +542,28 @@ impl Backend {
         predicate: &Predicate,
         deadline: Instant,
     ) -> Result<QueryRows> {
-        let selection = self.selected_objects(streams, window)?;
-        if selection.paths.is_empty() {
+        let permits_partition_scan =
+            !predicate.ids.is_empty() || !predicate.sessions.is_empty();
+        let (selection, partition_scan) = match self.selected_objects(streams, window) {
+            Ok(selection) => (selection, false),
+            Err(error)
+                if permits_partition_scan
+                    && matches!(
+                        error.downcast_ref::<TraceQueryError>(),
+                        Some(TraceQueryError::Limit(_))
+                    ) =>
+            {
+                (
+                    ObjectSelection {
+                        days: window_days(window).into_iter().collect(),
+                        paths: Vec::new(),
+                    },
+                    true,
+                )
+            }
+            Err(error) => return Err(error),
+        };
+        if selection.days.is_empty() || (!partition_scan && selection.paths.is_empty()) {
             metrics::Run::new("athena_trace")
                 .set("outcome", "empty")
                 .set("rows", 0)
@@ -751,7 +784,6 @@ fn select_sql(
         "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))
@@ -762,15 +794,19 @@ fn select_sql(
         .map(|day| sql_string(day))
         .collect::<Vec<_>>()
         .join(", ");
-    let path_values = paths
-        .iter()
-        .map(|path| sql_string(path))
-        .collect::<Vec<_>>()
-        .join(", ");
     let mut clauses = vec![
         format!("stream IN ({stream_values})"),
         format!("day IN ({day_values})"),
-        format!("\"$path\" IN ({path_values})"),
+    ];
+    if !paths.is_empty() {
+        let path_values = paths
+            .iter()
+            .map(|path| sql_string(path))
+            .collect::<Vec<_>>()
+            .join(", ");
+        clauses.push(format!("\"$path\" IN ({path_values})"));
+    }
+    clauses.extend([
         format!(
             "from_iso8601_timestamp(json_extract_scalar(line, '$.ts')) >= from_iso8601_timestamp({})",
             sql_string(&window.since.to_rfc3339())
@@ -779,7 +815,7 @@ fn select_sql(
             "from_iso8601_timestamp(json_extract_scalar(line, '$.ts')) < from_iso8601_timestamp({})",
             sql_string(&window.until.to_rfc3339())
         ),
-    ];
+    ]);
     if let Some(source) = predicate.source.as_deref() {
         clauses.push(format!(
             "strpos(lower(coalesce(json_extract_scalar(line, '$.source'), '')), {}) > 0",
@@ -881,6 +917,15 @@ fn event_sessions(lines: &[String]) -> BTreeSet<String> {
     sessions
 }
 
+fn event_streams(lines: &[String]) -> BTreeSet<String> {
+    lines
+        .iter()
+        .filter_map(|line| serde_json::from_str::<crate::event::Event>(line).ok())
+        .map(|event| event.stream)
+        .filter(|stream| !stream.is_empty())
+        .collect()
+}
+
 fn sql_string(value: &str) -> String {
     format!("'{}'", value.replace('\'', "''"))
 }
@@ -889,6 +934,30 @@ fn query_id(id: &str) -> &str {
     id.strip_prefix("job:").unwrap_or(id)
 }
 
+fn id_lookup_window(id: &str) -> Option<Window> {
+    ids_lookup_window(&[id])
+}
+
+fn ids_lookup_window(ids: &[&str]) -> Option<Window> {
+    let mut timestamps = ids.iter().map(|id| {
+        let timestamp = crate::event::ulid_timestamp_ms(query_id(id))?;
+        DateTime::<Utc>::from_timestamp_millis(timestamp as i64)
+    });
+    let first = timestamps.next()??;
+    let (mut since, mut until) = (first, first);
+    for timestamp in timestamps {
+        let timestamp = timestamp?;
+        since = since.min(timestamp);
+        until = until.max(timestamp);
+    }
+    let window = Window {
+        since: since - Duration::minutes(ID_LOOKUP_MINUTES),
+        until: until + Duration::minutes(ID_LOOKUP_MINUTES),
+    };
+    (window.until - window.since <= Duration::hours(MAX_LOOKBACK_HOURS))
+        .then_some(window)
+}
+
 fn normalized_stream_source(source: &str) -> String {
     match source.to_ascii_lowercase().as_str() {
         "codex" | "codex_cli" | "codex-cli" => "codex".into(),
@@ -1006,6 +1075,66 @@ mod tests {
         }
     }
 
+    #[test]
+    fn exact_id_falls_back_to_stream_day_partitions_when_paths_exceed_the_guard() {
+        use crate::bucket::Bucket;
+
+        let root = std::env::temp_dir().join(format!(
+            "synty-athena-id-partition-fallback-{}",
+            std::process::id()
+        ));
+        let _ = std::fs::remove_dir_all(&root);
+        let bucket = crate::bucket::LocalFs::new(&root);
+        for index in 0..=MAX_OBJECT_PATHS {
+            bucket
+                .put(
+                    &format!(
+                        "events/edge-m-codex/chunks/track.2026-07-22/{index:04}.jsonl"
+                    ),
+                    b"{}\n",
+                )
+                .unwrap();
+        }
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        let mut 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::clone(&calls),
+            }),
+            streams: Some(vec!["edge-m-codex".into()]),
+            days: None,
+            cached: None,
+        };
+
+        backend
+            .select(
+                &["edge-m-codex".into()],
+                Window {
+                    since: parse_time("2026-07-22T10:00:00Z").unwrap(),
+                    until: parse_time("2026-07-22T10:10:00Z").unwrap(),
+                },
+                &Predicate {
+                    ids: vec!["event-id".into()],
+                    ..Default::default()
+                },
+                Instant::now() + StdDuration::from_secs(5),
+            )
+            .unwrap();
+
+        let calls = calls.lock().unwrap();
+        assert_eq!(calls.len(), 1);
+        assert!(calls[0].contains("day IN ('2026-07-22')"));
+        assert!(!calls[0].contains("\"$path\""));
+        assert!(calls[0].contains("'event-id'"));
+        let _ = std::fs::remove_dir_all(&root);
+    }
+
     #[test]
     fn raw_athena_rows_reconstruct_the_existing_span_surface() {
         let lines = vec![
@@ -1182,6 +1311,73 @@ mod tests {
         assert!(!calls[1].contains("synty_context_rank"));
     }
 
+    #[test]
+    fn show_uses_ulid_time_and_matching_stream_for_bounded_lookup() {
+        let timestamp = parse_time("2026-07-22T10:00:01Z").unwrap();
+        let id = crate::event::deterministic_ulid(timestamp.timestamp_millis() as u64, "call");
+        let lines = vec![
+            event(
+                "start",
+                "2026-07-22T10:00:00Z",
+                "session_start",
+                json!({"cwd":"/work/synty"}),
+            ),
+            event(
+                &id,
+                "2026-07-22T10:00:01Z",
+                "tool_call",
+                json!({"name":"exec_command","call_id":"c1","arguments":"{\"cmd\":\"cargo test\"}"}),
+            ),
+        ];
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        let mut backend = Backend {
+            config: Config::new(
+                "s3://bucket".into(),
+                "wg".into(),
+                "synty".into(),
+                "raw_events".into(),
+            )
+            .unwrap(),
+            query: Box::new(FakeQuery {
+                lines,
+                calls: Arc::clone(&calls),
+            }),
+            streams: Some(vec![
+                "edge-m-codex".into(),
+                "edge-other-claudecode".into(),
+            ]),
+            days: Some(vec!["2026-07-22".into()]),
+            cached: None,
+        };
+
+        let out = backend
+            .show(&id, 3, 5, &ReadScope::default())
+            .unwrap();
+
+        assert!(out.contains("cargo test"));
+        let calls = calls.lock().unwrap();
+        assert_eq!(calls.len(), 2);
+        assert!(calls[0].contains("2026-07-22T09:55:01+00:00"));
+        assert!(calls[0].contains("2026-07-22T10:05:01+00:00"));
+        assert!(calls[0].contains("'edge-other-claudecode'"));
+        assert!(calls[1].contains("stream IN ('edge-m-codex')"));
+        assert!(!calls[1].contains("'edge-other-claudecode'"));
+    }
+
+    #[test]
+    fn compare_window_covers_both_ulid_timestamps() {
+        let early = parse_time("2026-07-22T10:00:00Z").unwrap();
+        let late = parse_time("2026-07-22T10:30:00Z").unwrap();
+        let left = crate::event::deterministic_ulid(early.timestamp_millis() as u64, "left");
+        let right = crate::event::deterministic_ulid(late.timestamp_millis() as u64, "right");
+
+        let window = ids_lookup_window(&[&left, &format!("job:{right}")]).unwrap();
+
+        assert_eq!(window.since, parse_time("2026-07-22T09:55:00Z").unwrap());
+        assert_eq!(window.until, parse_time("2026-07-22T10:35:00Z").unwrap());
+        assert!(ids_lookup_window(&[&left, "foreign-id"]).is_none());
+    }
+
     #[test]
     fn listed_job_ids_query_the_native_span_id() {
         let lines = vec![