From 9c6b7b4213297d68a6de4cae7ef6ec3aa8b8394c Mon Sep 17 00:00:00 2001
From: Attila Laszlo Nagy <attilalaszlo.nagy@superlinked.com>
Date: Fri, 24 Jul 2026 01:09:18 +0200
Subject: [PATCH 1/3] M10: Query remote traces through Athena

Remove the remote MCP dependency on a generated trace projection by cataloging the existing immutable event chunks with Glue partition projection and folding bounded SELECT results through the existing trace model and read-scope policy.

Live sie-gw-cfg validation returned four matching tool events and opened one as a paired 193 ms span. Kind pushdown reduced the match selection from 184 rows to 14, and metadata deduplication reduced its context selection to 2 rows. The 4.1 GiB legacy trace.json was removed from the PVC; authenticated search still succeeded afterward.

Keep every request within 7 days, 50,000 events, 64 MiB of returned envelopes, 50 seconds per SELECT, and a 20 GiB workgroup scan cutoff. The workload role retains only S3 Get/List plus workgroup-scoped Athena and Glue read-query actions.
---
 .github/workflows/ci.yml                 |   9 +-
 .github/workflows/release.yml            |   6 +-
 AGENTS.md                                |   4 +-
 Cargo.lock                               | 259 ++++++-
 Cargo.toml                               |   4 +
 Dockerfile                               |   4 +-
 README.md                                |  47 +-
 deploy/aws/athena-trace.yaml             | 109 +++
 deploy/aws/mcp-reader.yaml               |  28 +
 deploy/helm/synty/templates/mcp.yaml     |  31 +-
 deploy/helm/synty/values-sie-gw-cfg.yaml |  17 +
 deploy/helm/synty/values.yaml            |  10 +
 docs/design.md                           |  82 +-
 src/main.rs                              |  44 +-
 src/mcp.rs                               | 184 ++++-
 src/mcp_http.rs                          |  56 +-
 src/readmodel.rs                         |  12 +-
 src/sync.rs                              |  53 +-
 src/trace.rs                             | 109 ++-
 src/trace_athena.rs                      | 911 +++++++++++++++++++++++
 20 files changed, 1831 insertions(+), 148 deletions(-)
 create mode 100644 deploy/aws/athena-trace.yaml
 create mode 100644 src/trace_athena.rs

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index cdba193..3cbfa9c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -19,10 +19,10 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - run: cargo test --locked
       - run: cargo test --locked --no-default-features
-      - run: cargo test --locked --features s3,gcs,mcp-http
+      - run: cargo test --locked --features s3,gcs,mcp-http,athena
       - name: Exercise the HTTP MCP listener
         run: |
-          cargo build --locked --features s3,gcs,mcp-http
+          cargo build --locked --features s3,gcs,mcp-http,athena
           scripts/mcp-http-smoke.sh target/debug/synty
       - run: git diff --check
 
@@ -75,13 +75,18 @@ jobs:
           --set mcp.enabled=true
           --set mcp.tls.existingSecret=synty-mcp-tls
           --set mcp.allowedOrigins[0]=https://memory.example.com
+          --set mcp.athena.enabled=true
+          --set mcp.athena.workgroup=synty-mcp-readonly
           --set-json 'networkPolicy.objectStoreCidrs=["203.0.113.10/32"]'
+          --set-json 'networkPolicy.awsApiCidrs=["198.51.100.20/32"]'
           --set persistence.enabled=false >/tmp/synty-options.yaml
       - name: Assert MCP probe and egress contracts
         run: |
           grep -q 'path: /ready' /tmp/synty-options.yaml
           grep -q 'scheme: HTTPS' /tmp/synty-options.yaml
           grep -q '203.0.113.10/32' /tmp/synty-options.yaml
+          grep -q -- '--athena-workgroup synty-mcp-readonly' /tmp/synty-options.yaml
+          grep -q '198.51.100.20/32' /tmp/synty-options.yaml
           if grep -q -- '- {}' /tmp/synty-options.yaml; then
             echo "MCP rendered an unrestricted NetworkPolicy rule" >&2
             exit 1
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 021d719..5ca8502 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -19,7 +19,7 @@ jobs:
       - uses: dtolnay/rust-toolchain@1.97.0
       - uses: Swatinem/rust-cache@v2
       - run: cargo test --locked # pure: no model, corpus, or network
-      - run: cargo test --locked --features s3,gcs,mcp-http # compile + test shipped remote paths
+      - run: cargo test --locked --features s3,gcs,mcp-http,athena # compile + test shipped remote paths
       - name: Verify tag, binary, and chart versions agree
         run: |
           cargo_version="$(awk -F ' *= *' '$1 == "version" {gsub(/"/, "", $2); print $2; exit}' Cargo.toml)"
@@ -38,11 +38,11 @@ jobs:
           # Apple Silicon: GPU encode via metal, with runtime CPU fallback.
           - os: macos-14
             asset: synty-darwin-arm64
-            features: metal,s3,gcs,mcp-http
+            features: metal,s3,gcs,mcp-http,athena
           # Linux x64: plain CPU, portable.
           - os: ubuntu-latest
             asset: synty-linux-x64
-            features: s3,gcs,mcp-http
+            features: s3,gcs,mcp-http,athena
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
diff --git a/AGENTS.md b/AGENTS.md
index 6c03afe..bb94bf4 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -33,8 +33,8 @@ Rust binary.
   the other opt-in backends. None may become default.
 - Distribution is GitHub Releases, cut by `.github/workflows/release.yml` on a
   `v*` tag: it builds per platform with explicit features (this does not change
-  the default above) — `--features metal,s3,gcs,mcp-http` for the macOS asset,
-  `--features s3,gcs,mcp-http` for Linux — and attaches `synty-<os>-<arch>`
+  the default above) — `--features metal,s3,gcs,mcp-http,athena` for the macOS asset,
+  `--features s3,gcs,mcp-http,athena` for Linux — and attaches `synty-<os>-<arch>`
   (+ `.sha256`).
   `synty upgrade` self-updates from the latest release (sha256-verified, via the
   GitHub token); a cached nag flags when behind. These are ops, not pipeline
diff --git a/Cargo.lock b/Cargo.lock
index a84b451..dc5dcf8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -224,6 +224,28 @@ dependencies = [
  "uuid",
 ]
 
+[[package]]
+name = "aws-sdk-athena"
+version = "1.84.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "84dddffcffec69e8a75dcdf52dd7a31f5a87b4c7948909d3f3b0991a16382cb6"
+dependencies = [
+ "aws-credential-types",
+ "aws-runtime",
+ "aws-smithy-async",
+ "aws-smithy-http",
+ "aws-smithy-json",
+ "aws-smithy-runtime",
+ "aws-smithy-runtime-api",
+ "aws-smithy-types",
+ "aws-types",
+ "bytes",
+ "fastrand",
+ "http 0.2.12",
+ "regex-lite",
+ "tracing",
+]
+
 [[package]]
 name = "aws-sdk-sts"
 version = "1.89.0"
@@ -310,17 +332,23 @@ dependencies = [
  "aws-smithy-async",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
- "h2",
+ "h2 0.3.27",
+ "h2 0.4.14",
+ "http 0.2.12",
  "http 1.4.1",
- "hyper",
- "hyper-rustls",
+ "http-body 0.4.6",
+ "hyper 0.14.32",
+ "hyper 1.10.1",
+ "hyper-rustls 0.24.2",
+ "hyper-rustls 0.27.9",
  "hyper-util",
  "pin-project-lite",
- "rustls",
- "rustls-native-certs",
+ "rustls 0.21.12",
+ "rustls 0.23.40",
+ "rustls-native-certs 0.8.3",
  "rustls-pki-types",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.26.4",
  "tower",
  "tracing",
 ]
@@ -403,6 +431,7 @@ dependencies = [
  "base64-simd",
  "bytes",
  "bytes-utils",
+ "futures-core",
  "http 0.2.12",
  "http 1.4.1",
  "http-body 0.4.6",
@@ -415,6 +444,8 @@ dependencies = [
  "ryu",
  "serde",
  "time",
+ "tokio",
+ "tokio-util",
 ]
 
 [[package]]
@@ -452,6 +483,12 @@ version = "0.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8"
 
+[[package]]
+name = "base64"
+version = "0.21.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567"
+
 [[package]]
 name = "base64"
 version = "0.22.1"
@@ -1740,6 +1777,25 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "h2"
+version = "0.3.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0beca50380b1fc32983fc1cb4587bfa4bb9e78fc259aad4a0032d2080309222d"
+dependencies = [
+ "bytes",
+ "fnv",
+ "futures-core",
+ "futures-sink",
+ "futures-util",
+ "http 0.2.12",
+ "indexmap",
+ "slab",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
 [[package]]
 name = "h2"
 version = "0.4.14"
@@ -1933,6 +1989,30 @@ version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424"
 
+[[package]]
+name = "hyper"
+version = "0.14.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "41dfc780fdec9373c01bae43289ea34c972e40ee3c9f6b3c8801a35f35586ce7"
+dependencies = [
+ "bytes",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "h2 0.3.27",
+ "http 0.2.12",
+ "http-body 0.4.6",
+ "httparse",
+ "httpdate",
+ "itoa",
+ "pin-project-lite",
+ "socket2 0.5.10",
+ "tokio",
+ "tower-service",
+ "tracing",
+ "want",
+]
+
 [[package]]
 name = "hyper"
 version = "1.10.1"
@@ -1943,7 +2023,7 @@ dependencies = [
  "bytes",
  "futures-channel",
  "futures-core",
- "h2",
+ "h2 0.4.14",
  "http 1.4.1",
  "http-body 1.0.1",
  "httparse",
@@ -1955,6 +2035,22 @@ dependencies = [
  "want",
 ]
 
+[[package]]
+name = "hyper-rustls"
+version = "0.24.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590"
+dependencies = [
+ "futures-util",
+ "http 0.2.12",
+ "hyper 0.14.32",
+ "log",
+ "rustls 0.21.12",
+ "rustls-native-certs 0.6.3",
+ "tokio",
+ "tokio-rustls 0.24.1",
+]
+
 [[package]]
 name = "hyper-rustls"
 version = "0.27.9"
@@ -1962,12 +2058,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f"
 dependencies = [
  "http 1.4.1",
- "hyper",
+ "hyper 1.10.1",
  "hyper-util",
- "rustls",
- "rustls-native-certs",
+ "rustls 0.23.40",
+ "rustls-native-certs 0.8.3",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.26.4",
  "tower-service",
  "webpki-roots 1.0.7",
 ]
@@ -1984,12 +2080,12 @@ dependencies = [
  "futures-util",
  "http 1.4.1",
  "http-body 1.0.1",
- "hyper",
+ "hyper 1.10.1",
  "ipnet",
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2",
+ "socket2 0.6.4",
  "tokio",
  "tower-service",
  "tracing",
@@ -2743,7 +2839,7 @@ dependencies = [
  "chrono",
  "futures",
  "humantime",
- "hyper",
+ "hyper 1.10.1",
  "itertools 0.13.0",
  "md-5",
  "parking_lot",
@@ -2752,7 +2848,7 @@ dependencies = [
  "rand 0.8.6",
  "reqwest",
  "ring",
- "rustls-pemfile",
+ "rustls-pemfile 2.2.0",
  "serde",
  "serde_json",
  "snafu",
@@ -2838,6 +2934,12 @@ dependencies = [
  "pkg-config",
 ]
 
+[[package]]
+name = "openssl-probe"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e"
+
 [[package]]
 name = "openssl-probe"
 version = "0.2.1"
@@ -3125,8 +3227,8 @@ dependencies = [
  "quinn-proto",
  "quinn-udp",
  "rustc-hash",
- "rustls",
- "socket2",
+ "rustls 0.23.40",
+ "socket2 0.6.4",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
@@ -3145,7 +3247,7 @@ dependencies = [
  "rand 0.9.4",
  "ring",
  "rustc-hash",
- "rustls",
+ "rustls 0.23.40",
  "rustls-pki-types",
  "slab",
  "thiserror 2.0.18",
@@ -3163,7 +3265,7 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2",
+ "socket2 0.6.4",
  "tracing",
  "windows-sys 0.60.2",
 ]
@@ -3417,27 +3519,27 @@ dependencies = [
  "bytes",
  "futures-core",
  "futures-util",
- "h2",
+ "h2 0.4.14",
  "http 1.4.1",
  "http-body 1.0.1",
  "http-body-util",
- "hyper",
- "hyper-rustls",
+ "hyper 1.10.1",
+ "hyper-rustls 0.27.9",
  "hyper-util",
  "js-sys",
  "log",
  "percent-encoding",
  "pin-project-lite",
  "quinn",
- "rustls",
- "rustls-native-certs",
+ "rustls 0.23.40",
+ "rustls-native-certs 0.8.3",
  "rustls-pki-types",
  "serde",
  "serde_json",
  "serde_urlencoded",
  "sync_wrapper",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.26.4",
  "tokio-util",
  "tower",
  "tower-http",
@@ -3530,6 +3632,18 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "rustls"
+version = "0.21.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e"
+dependencies = [
+ "log",
+ "ring",
+ "rustls-webpki 0.101.7",
+ "sct",
+]
+
 [[package]]
 name = "rustls"
 version = "0.23.40"
@@ -3541,21 +3655,42 @@ dependencies = [
  "once_cell",
  "ring",
  "rustls-pki-types",
- "rustls-webpki",
+ "rustls-webpki 0.103.13",
  "subtle",
  "zeroize",
 ]
 
+[[package]]
+name = "rustls-native-certs"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00"
+dependencies = [
+ "openssl-probe 0.1.6",
+ "rustls-pemfile 1.0.4",
+ "schannel",
+ "security-framework 2.11.1",
+]
+
 [[package]]
 name = "rustls-native-certs"
 version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
 dependencies = [
- "openssl-probe",
+ "openssl-probe 0.2.1",
  "rustls-pki-types",
  "schannel",
- "security-framework",
+ "security-framework 3.7.0",
+]
+
+[[package]]
+name = "rustls-pemfile"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1c74cae0a4cf6ccbbf5f359f08efdf8ee7e1dc532573bf0db71968cb56b1448c"
+dependencies = [
+ "base64 0.21.7",
 ]
 
 [[package]]
@@ -3577,6 +3712,16 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "rustls-webpki"
+version = "0.101.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b6275d1ee7a1cd780b64aca7726599a1dbc893b1e64144529e55c3c2f745765"
+dependencies = [
+ "ring",
+ "untrusted",
+]
+
 [[package]]
 name = "rustls-webpki"
 version = "0.103.13"
@@ -3656,6 +3801,29 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
+[[package]]
+name = "sct"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da046153aa2352493d6cb7da4b6e5c0c057d8a1d0a9aa8560baffdd945acd414"
+dependencies = [
+ "ring",
+ "untrusted",
+]
+
+[[package]]
+name = "security-framework"
+version = "2.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02"
+dependencies = [
+ "bitflags 2.11.1",
+ "core-foundation 0.9.4",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
 [[package]]
 name = "security-framework"
 version = "3.7.0"
@@ -3851,6 +4019,16 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.5.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678"
+dependencies = [
+ "libc",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "socket2"
 version = "0.6.4"
@@ -3981,6 +4159,7 @@ dependencies = [
  "async-trait",
  "aws-config",
  "aws-credential-types",
+ "aws-sdk-athena",
  "bytes",
  "candle-core",
  "candle-nn",
@@ -3991,7 +4170,7 @@ dependencies = [
  "futures",
  "half",
  "http-body-util",
- "hyper",
+ "hyper 1.10.1",
  "hyper-util",
  "libc",
  "ndarray",
@@ -4000,14 +4179,14 @@ dependencies = [
  "pylate-rs",
  "ratatui",
  "regex",
- "rustls",
- "rustls-pemfile",
+ "rustls 0.23.40",
+ "rustls-pemfile 2.2.0",
  "serde",
  "serde_json",
  "sha2",
  "tokenizers",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.26.4",
  "ureq",
  "walkdir",
 ]
@@ -4177,7 +4356,7 @@ dependencies = [
  "mio",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "socket2 0.6.4",
  "tokio-macros",
  "windows-sys 0.61.2",
 ]
@@ -4193,13 +4372,23 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "tokio-rustls"
+version = "0.24.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081"
+dependencies = [
+ "rustls 0.21.12",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-rustls"
 version = "0.26.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61"
 dependencies = [
- "rustls",
+ "rustls 0.23.40",
  "tokio",
 ]
 
@@ -4464,7 +4653,7 @@ dependencies = [
  "flate2",
  "log",
  "once_cell",
- "rustls",
+ "rustls 0.23.40",
  "rustls-pki-types",
  "serde",
  "serde_json",
diff --git a/Cargo.toml b/Cargo.toml
index b5059a1..39d3750 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -36,6 +36,9 @@ s3 = [
     "dep:async-trait",
 ]
 gcs = ["dep:object_store", "object_store/gcp", "dep:tokio", "dep:futures"]
+# Bounded, read-only raw-event queries for remote trace tools. This stays
+# opt-in so the default local binary remains dependency-light.
+athena = ["s3", "dep:aws-sdk-athena"]
 # Authenticated Streamable HTTP transport for remote MCP clients. Stdio stays
 # available in every build; release/container artifacts opt into this feature.
 mcp-http = [
@@ -86,6 +89,7 @@ futures = { version = "0.3", optional = true }
 # releases raise their MSRV independently of Synty.
 aws-config = { version = "=1.8.8", optional = true, default-features = false, features = ["rt-tokio", "credentials-process", "default-https-client"] }
 aws-credential-types = { version = "=1.2.8", optional = true }
+aws-sdk-athena = { version = "=1.84.0", optional = true, default-features = false, features = ["default-https-client", "rt-tokio", "rustls"] }
 async-trait = { version = "0.1", optional = true }
 bytes = { version = "1", optional = true }
 http-body-util = { version = "0.1", optional = true }
diff --git a/Dockerfile b/Dockerfile
index a836e21..53740f1 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -8,13 +8,13 @@ COPY Cargo.toml Cargo.lock ./
 # the expensive native dependencies on both architecture runners.
 RUN mkdir src \
     && printf 'fn main() {}\n' > src/main.rs \
-    && cargo build --release --locked --features s3,gcs,mcp-http \
+    && cargo build --release --locked --features s3,gcs,mcp-http,athena \
     && rm -rf src
 COPY src ./src
 # Docker normalizes copied mtimes; force Cargo to invalidate the dummy package
 # while retaining every dependency artifact from the preceding layer.
 RUN touch src/main.rs \
-    && cargo build --release --locked --features s3,gcs,mcp-http
+    && cargo build --release --locked --features s3,gcs,mcp-http,athena
 
 FROM debian:bookworm-slim
 
diff --git a/README.md b/README.md
index c813219..2e2bc9f 100644
--- a/README.md
+++ b/README.md
@@ -115,6 +115,7 @@ synty import run.ndjson --format harness --machine eval-1 --campaign c42 \
 # Streamable HTTP; a non-loopback bind additionally requires HTTPS material.
 synty mcp --role investigator --scope scope.json --redaction mcp_safe
 SYNTY_MCP_TOKEN="$token" synty mcp --http --bind 0.0.0.0:8765 \
+  --bucket s3://my-team-synty --athena-workgroup synty-mcp-readonly \
   --listen-public --tls-cert tls.crt --tls-key tls.key \
   --allowed-origin https://memory.example.com --scope scope.json
 ```
@@ -168,11 +169,15 @@ tracker starts at boot without an SSH login, then run `init` normally:
 sudo loginctl enable-linger "$USER"
 ```
 
-That bucket is the only shared infrastructure: no build server, no coordination
-service. Each machine writes a stable `edge-<machine>-<source>` stream, so
+The bucket is the durable shared backplane; an S3 deployment may optionally add
+Glue catalog metadata and a bounded Athena workgroup for remote trace queries.
+There is still no migration, crawler, build server, or coordination service.
+Each machine writes a stable `edge-<machine>-<source>` stream, so
 writers do not overwrite one another. Local readers and builders pull every
-stream plus the latest published read-model. MCP-only readers pull just that
-read-model, which includes compact session, tool, fleet, and trace projections.
+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;
 `synty build` does the same explicitly, while `search` warns if raw events are
@@ -198,10 +203,12 @@ For S3, scope each writer/reader role to the chosen bucket (or URI prefix):
 `s3:DeleteObject` on its objects. Delete is used only to release the soft build
 lease; event chunks and content-addressed derived objects are immutable.
 An MCP-only workload needs no writes: `deploy/aws/mcp-reader.yaml` creates an
-IRSA role with only bucket location/list and object get permissions. Use it
-with the tracker and builder disabled. If EKS and S3 are in different regions,
-set Helm's `bucketRegion` to the bucket's region so requests are signed for the
-correct endpoint.
+IRSA role with bucket location/list/object-get plus query-only access to one
+Athena workgroup and one Glue table. `deploy/aws/athena-trace.yaml` creates that
+external table and a managed-results workgroup without writing, crawling, or
+rewriting bucket objects. Use it with the tracker and builder disabled. If EKS
+and S3 are in different regions, set Helm's `bucketRegion` to the bucket's
+region so S3 and Athena requests are signed there.
 
 A one-shot local projection migration can still reuse that bucket's existing
 fleet embeddings without broadening the reader role:
@@ -227,11 +234,16 @@ name native producers such as `harness`, `codex_cli`, or `github`. Restricted
 scopes omit fleet-wide status/stat/tool surfaces and rebuild topic facets only
 from allowed members. HTTP `synty_related` accepts client-supplied `context`
 and never reads a caller-provided path on the server.
-MCP pulls the complete published read-model before serving and refreshes it on
-a background thread; it never mirrors the raw event lake. Format-2 builds carry
-compact session/tool/fleet facts plus a trace projection whose searchable
-evidence, commands, and outputs are capped per event. `/health` reports
-transport liveness and `/ready` waits for both projections. Analysis tools are
+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.
+`/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
 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,
@@ -248,8 +260,9 @@ tags build each architecture on a native GitHub runner, then publish a verified
 `deploy/aws/ecr-publisher.yaml` once and set the repository variable
 `AWS_ECR_PUBLISH_ROLE_ARN` to its role output. Remote MCP is disabled by default;
 enabling it requires a TLS Secret and a NetworkPolicy whose source selectors
-name trusted callers. The policy also requires explicit object-store CIDRs;
-cluster DNS and only those destinations on TCP 443 are allowed for egress.
+name trusted callers. The policy also requires explicit object-store CIDRs and,
+in Athena mode, regional AWS API CIDRs; cluster DNS and only those destinations
+on TCP 443 are allowed for egress.
 CIDR configuration is capped at 64 ranges and rejects IPv4 prefixes broader
 than `/12` or IPv6 prefixes broader than `/32`, including split-default-route
 combinations. The Service remains cluster-internal, and its default ingress
@@ -280,12 +293,12 @@ on-real-data validation lives in [`evals/`](evals/).
 
 ```sh
 cargo build --release                          # plain CPU, portable (the shipped core)
-cargo build --release --features s3,gcs,mcp-http   # team + remote MCP container build
+cargo build --release --features s3,gcs,mcp-http,athena   # team + remote MCP container build
 ```
 
 On Apple Silicon, add `--features metal` for GPU encode (~5.7× faster);
 `accelerate` (macOS) and `mkl` (Linux) are CPU-BLAS alternatives. Release assets
-and the Dockerfile include `mcp-http`. The embedding model (~127 MB) downloads
+and the Dockerfile include `mcp-http` and `athena`. The embedding model (~127 MB) downloads
 on first use; the summarizer (~1.2 GB) the first time anything summarizes. For
 an air-gapped setup and the per-stage pipeline, see
 [`docs/design.md`](docs/design.md) and [CONTRIBUTING](CONTRIBUTING.md).
diff --git a/deploy/aws/athena-trace.yaml b/deploy/aws/athena-trace.yaml
new file mode 100644
index 0000000..5b31128
--- /dev/null
+++ b/deploy/aws/athena-trace.yaml
@@ -0,0 +1,109 @@
+AWSTemplateFormatVersion: "2010-09-09"
+Description: Read-only Glue catalog and bounded Athena workgroup for Synty raw event traces
+
+Parameters:
+  BucketName:
+    Type: String
+  DatabaseName:
+    Type: String
+    Default: synty
+    AllowedPattern: "[a-z0-9_]+"
+  TableName:
+    Type: String
+    Default: raw_events
+    AllowedPattern: "[a-z0-9_]+"
+  WorkGroupName:
+    Type: String
+    Default: synty-mcp-readonly
+    AllowedPattern: "[a-zA-Z0-9._-]+"
+  FirstEventDay:
+    Type: String
+    Default: "2026-01-01"
+    AllowedPattern: "[0-9]{4}-[0-9]{2}-[0-9]{2}"
+  BytesScannedCutoffPerQuery:
+    Type: Number
+    Default: 21474836480
+    MinValue: 10000000
+    Description: Hard per-query scan ceiling; default is 20 GiB.
+
+Resources:
+  TraceDatabase:
+    Type: AWS::Glue::Database
+    Properties:
+      CatalogId: !Ref AWS::AccountId
+      DatabaseInput:
+        Name: !Ref DatabaseName
+        Description: Synty read-only raw event catalog
+
+  RawEventsTable:
+    Type: AWS::Glue::Table
+    Properties:
+      CatalogId: !Ref AWS::AccountId
+      DatabaseName: !Ref TraceDatabase
+      TableInput:
+        Name: !Ref TableName
+        Description: Immutable Synty JSONL event envelopes; one raw line per row
+        Owner: synty
+        TableType: EXTERNAL_TABLE
+        Parameters:
+          EXTERNAL: "TRUE"
+          classification: regex
+          projection.enabled: "true"
+          projection.stream.type: injected
+          projection.day.type: date
+          projection.day.format: yyyy-MM-dd
+          projection.day.interval: "1"
+          projection.day.interval.unit: DAYS
+          projection.day.range: !Sub "${FirstEventDay},NOW"
+          storage.location.template: !Sub
+            - "s3://${Bucket}/events/${!stream}/chunks/track.${!day}/"
+            - Bucket: !Ref BucketName
+        PartitionKeys:
+          - Name: stream
+            Type: string
+          - Name: day
+            Type: string
+        StorageDescriptor:
+          Columns:
+            - Name: line
+              Type: string
+              Comment: Complete immutable event envelope
+          Compressed: false
+          InputFormat: org.apache.hadoop.mapred.TextInputFormat
+          Location: !Sub "s3://${BucketName}/events/"
+          OutputFormat: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
+          SerdeInfo:
+            Name: synty-json-line
+            SerializationLibrary: org.apache.hadoop.hive.serde2.RegexSerDe
+            Parameters:
+              input.regex: "^(.*)$"
+
+  TraceWorkGroup:
+    Type: AWS::Athena::WorkGroup
+    Properties:
+      Name: !Ref WorkGroupName
+      Description: Read-only bounded queries over Synty raw event envelopes
+      State: ENABLED
+      RecursiveDeleteOption: true
+      WorkGroupConfiguration:
+        EnforceWorkGroupConfiguration: true
+        PublishCloudWatchMetricsEnabled: true
+        RequesterPaysEnabled: false
+        BytesScannedCutoffPerQuery: !Ref BytesScannedCutoffPerQuery
+        ManagedQueryResultsConfiguration:
+          Enabled: true
+        EngineVersion:
+          SelectedEngineVersion: AUTO
+      Tags:
+        - Key: application
+          Value: synty
+        - Key: access
+          Value: read-only
+
+Outputs:
+  DatabaseName:
+    Value: !Ref TraceDatabase
+  TableName:
+    Value: !Ref RawEventsTable
+  WorkGroupName:
+    Value: !Ref TraceWorkGroup
diff --git a/deploy/aws/mcp-reader.yaml b/deploy/aws/mcp-reader.yaml
index daaef92..74df895 100644
--- a/deploy/aws/mcp-reader.yaml
+++ b/deploy/aws/mcp-reader.yaml
@@ -18,6 +18,18 @@ Parameters:
   RoleName:
     Type: String
     Default: synty-mcp-reader
+  AthenaRegion:
+    Type: String
+    Default: us-west-2
+  AthenaWorkGroup:
+    Type: String
+    Default: synty-mcp-readonly
+  GlueDatabase:
+    Type: String
+    Default: synty
+  GlueTable:
+    Type: String
+    Default: raw_events
 
 Resources:
   ReaderRole:
@@ -53,6 +65,22 @@ Resources:
                   - s3:GetObject
                   - s3:GetObjectVersion
                 Resource: !Sub arn:${AWS::Partition}:s3:::${BucketName}/*
+              - Effect: Allow
+                Action:
+                  - athena:GetQueryExecution
+                  - athena:GetQueryResults
+                  - athena:StartQueryExecution
+                  - athena:StopQueryExecution
+                Resource: !Sub arn:${AWS::Partition}:athena:${AthenaRegion}:${AWS::AccountId}:workgroup/${AthenaWorkGroup}
+              - Effect: Allow
+                Action:
+                  - glue:GetDatabase
+                  - glue:GetTable
+                  - glue:GetPartitions
+                Resource:
+                  - !Sub arn:${AWS::Partition}:glue:${AthenaRegion}:${AWS::AccountId}:catalog
+                  - !Sub arn:${AWS::Partition}:glue:${AthenaRegion}:${AWS::AccountId}:database/${GlueDatabase}
+                  - !Sub arn:${AWS::Partition}:glue:${AthenaRegion}:${AWS::AccountId}:table/${GlueDatabase}/${GlueTable}
 
 Outputs:
   ReaderRoleArn:
diff --git a/deploy/helm/synty/templates/mcp.yaml b/deploy/helm/synty/templates/mcp.yaml
index bc20e2d..488ae19 100644
--- a/deploy/helm/synty/templates/mcp.yaml
+++ b/deploy/helm/synty/templates/mcp.yaml
@@ -11,18 +11,27 @@
 {{- if gt (len .Values.networkPolicy.objectStoreCidrs) 64 }}
 {{- fail "networkPolicy.objectStoreCidrs must contain at most 64 endpoint ranges" }}
 {{- end }}
-{{- range .Values.networkPolicy.objectStoreCidrs }}
+{{- if and .Values.mcp.athena.enabled (not .Values.mcp.athena.workgroup) }}
+{{- fail "mcp.athena.workgroup is required when Athena trace is enabled" }}
+{{- end }}
+{{- if and .Values.mcp.athena.enabled (not .Values.networkPolicy.awsApiCidrs) }}
+{{- fail "networkPolicy.awsApiCidrs must list Athena and credential endpoint ranges when Athena trace is enabled" }}
+{{- end }}
+{{- if gt (len .Values.networkPolicy.awsApiCidrs) 64 }}
+{{- fail "networkPolicy.awsApiCidrs must contain at most 64 endpoint ranges" }}
+{{- end }}
+{{- range concat .Values.networkPolicy.objectStoreCidrs .Values.networkPolicy.awsApiCidrs }}
 {{- if not (regexMatch `^.+/[0-9]+$` .) }}
-{{- fail "networkPolicy.objectStoreCidrs entries must be CIDR ranges" }}
+{{- fail "networkPolicy endpoint entries must be CIDR ranges" }}
 {{- end }}
 {{- $prefix := atoi (last (splitList "/" .)) }}
 {{- if contains ":" . }}
 {{- if or (lt $prefix 32) (gt $prefix 128) }}
-{{- fail "networkPolicy.objectStoreCidrs IPv6 ranges must be /32 or narrower" }}
+{{- fail "networkPolicy endpoint IPv6 ranges must be /32 or narrower" }}
 {{- end }}
 {{- else }}
 {{- if or (lt $prefix 12) (gt $prefix 32) }}
-{{- fail "networkPolicy.objectStoreCidrs IPv4 ranges must be /12 or narrower" }}
+{{- fail "networkPolicy endpoint IPv4 ranges must be /12 or narrower" }}
 {{- end }}
 {{- end }}
 {{- end }}
@@ -83,6 +92,9 @@ spec:
           args:
             - |
               set -- synty mcp --bucket "$SYNTY_BUCKET" --http --listen-public --bind "0.0.0.0:{{ .Values.mcp.port }}" --tls-cert /etc/synty/tls/tls.crt --tls-key /etc/synty/tls/tls.key --role "$SYNTY_MCP_ROLE" --scope /etc/synty/scope.json --redaction mcp_safe
+              {{- if .Values.mcp.athena.enabled }}
+              set -- "$@" --athena-workgroup {{ .Values.mcp.athena.workgroup | quote }} --athena-database {{ .Values.mcp.athena.database | quote }} --athena-table {{ .Values.mcp.athena.table | quote }}
+              {{- end }}
               {{- range .Values.mcp.allowedOrigins }}
               set -- "$@" --allowed-origin {{ . | quote }}
               {{- end }}
@@ -107,10 +119,9 @@ spec:
             - { name: tmp, mountPath: /tmp }
             - { name: scope, mountPath: /etc/synty, readOnly: true }
             - { name: tls, mountPath: /etc/synty/tls, readOnly: true }
-          # Liveness proves the transport loop is responsive. Readiness also
-          # requires the compact analysis + trace projections from a format-2
-          # published build, so agents never discover a reader that must scan
-          # the raw lake on its first request.
+          # Liveness proves the transport loop is responsive. Readiness
+          # requires the semantic index and compact analysis projection; local
+          # trace mode also requires trace.json, while Athena mode does not.
           startupProbe:
             httpGet: { path: /health, port: mcp, scheme: HTTPS }
             periodSeconds: 10
@@ -185,9 +196,9 @@ spec:
       ports:
         - { protocol: UDP, port: 53 }
         - { protocol: TCP, port: 53 }
-    # Reach only the configured object-store address ranges over TLS.
+    # Reach only the configured object-store and AWS API address ranges over TLS.
     - to:
-        {{- range .Values.networkPolicy.objectStoreCidrs }}
+        {{- range concat .Values.networkPolicy.objectStoreCidrs .Values.networkPolicy.awsApiCidrs }}
         - ipBlock: { cidr: {{ . | quote }} }
         {{- end }}
       ports:
diff --git a/deploy/helm/synty/values-sie-gw-cfg.yaml b/deploy/helm/synty/values-sie-gw-cfg.yaml
index 7333b0b..185aded 100644
--- a/deploy/helm/synty/values-sie-gw-cfg.yaml
+++ b/deploy/helm/synty/values-sie-gw-cfg.yaml
@@ -24,6 +24,11 @@ builder:
 mcp:
   enabled: true
   role: operator
+  athena:
+    enabled: true
+    workgroup: synty-mcp-readonly
+    database: synty
+    table: raw_events
   tls:
     existingSecret: synty-mcp-tls
   # Empty allowlists deliberately expose all internal bucket records. The MCP
@@ -60,3 +65,15 @@ networkPolicy:
     - 3.2.68.0/24
     - 35.80.36.208/28
     - 35.80.36.224/28
+  # Current public Athena us-west-2 answers plus regional STS for IRSA. Keep
+  # these /32s fail-closed and refresh them when AWS rotates endpoint DNS.
+  awsApiCidrs:
+    - 52.24.182.208/32
+    - 32.186.145.128/32
+    - 35.82.214.87/32
+    - 52.43.188.237/32
+    - 32.184.245.105/32
+    - 184.34.144.212/32
+    - 44.227.85.44/32
+    - 44.239.29.42/32
+    - 44.248.101.77/32
diff --git a/deploy/helm/synty/values.yaml b/deploy/helm/synty/values.yaml
index 444f4af..39e49e4 100644
--- a/deploy/helm/synty/values.yaml
+++ b/deploy/helm/synty/values.yaml
@@ -77,6 +77,13 @@ mcp:
     existingSecret: ""
     certificateKey: tls.crt
     privateKeyKey: tls.key
+  athena:
+    # Remote trace tools can query immutable raw events directly instead of
+    # requiring a generated trace.json on the MCP volume.
+    enabled: false
+    workgroup: ""
+    database: synty
+    table: raw_events
   scope:
     repos: []
     campaigns: []
@@ -98,3 +105,6 @@ networkPolicy:
   # object-store endpoint. At most 64 ranges are accepted; IPv4 ranges must be
   # /12 or narrower and IPv6 ranges /32 or narrower.
   objectStoreCidrs: []
+  # Additional regional AWS API endpoint addresses used by Athena and the
+  # workload identity credential chain. Required when mcp.athena.enabled.
+  awsApiCidrs: []
diff --git a/docs/design.md b/docs/design.md
index 1974b62..2e2bad5 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -36,6 +36,38 @@ data.
 - **Agents are first-class readers.** The primary agent surface is a CLI that
   prints Markdown to stdout; humans get a TUI over the same data.
 
+## Team query architecture
+
+```mermaid
+flowchart LR
+    W["Trackers and harness import"] -->|"immutable JSONL chunks"| S3["S3 raw event lake"]
+    S3 --> B["Builder"]
+    B --> E["Content-addressed embeddings"]
+    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"]
+    G --> A["Bounded Athena SELECT"]
+    A -->|"bounded raw envelopes"| F["Rust trace fold"]
+    F --> MCP
+    MCP --> H["Harness agents"]
+```
+
+Semantic and forensic reads deliberately take different paths. Search and
+related-work use the published next-plaid index; `embeddings/` remains a
+builder-side content-addressed cache and is not downloaded per MCP query.
+Remote `synty_trace_*` tools use Athena to prune immutable raw envelopes by
+stream and day, then reuse the normal Rust fold to derive turns, spans, and
+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.
+
 ## Engine
 
 - **Encode:** `pylate-rs` (ColBERT on Candle, ModernBERT backend). Default
@@ -180,11 +212,17 @@ runs on CI or a server without a developer machine.
   `--tls-key`. `--scope`, `--redaction`, and repeatable `--allowed-origin`
   configure the mediated boundary. The transport validates exact browser
   origins and protocol versions, bounds requests, and
-  keeps health responsive while tool calls run. It pulls the published query
-  model before serving and refreshes only that model in the background; it does
-  not mirror the raw event lake. Format-2 builds include compact session/tool/
-  fleet facts and bounded trace evidence. `/health` remains a liveness check,
-  while `/ready` requires both mediated projections. Analysis calls use a
+  keeps health responsive while tool calls run. It pulls the published semantic
+  index and compact analysis projection before serving and refreshes only those
+  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.
+  `/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
   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
@@ -230,11 +268,13 @@ a long wait. A continuation without a captured initiating call remains visible
 and is labeled `continuation_only` rather than guessed onto another command.
 Every duration says whether the source reported it or it is merely an event
 gap; the latter may include human approval or idle time and is deliberately not
-presented as tool runtime. Ingest regenerates the projection from JSONL and
-publishes it inside the immutable read-model. Event search evidence is capped at
-512 characters and rendered commands/outputs at 360, which makes query memory
-depend on a compact event vocabulary rather than retained raw bytes. The raw
-envelopes remain authoritative and lossless for local rebuilds. *Built.*
+presented as tool runtime. Local builds retain the compact projection for
+offline compatibility. Remote S3 MCP uses the Glue/Athena raw-event path
+instead: the external table exposes each JSONL envelope as one `line` column
+and partition projection maps `(stream, day)` directly onto existing keys.
+There is no one-time data migration or crawler. Returned rows are folded into
+the same trace structures, then the existing scope and rendering logic runs.
+The raw envelopes remain authoritative and lossless. *Built.*
 
 ## Tiers and the trust boundary
 
@@ -305,7 +345,8 @@ builds/<build>.<rev>.json              manifest: filename → blob, per (build,
 current.json                           the pointer, PUT last; readers never see
                                         a torn build; rev versions the clusters;
                                         format 2 includes analysis.json and
-                                        trace.json in each immutable build
+                                        a local-compatible trace.json, which
+                                        Athena MCP readers omit when pulling
 lease/build                            soft TTL lease electing one index builder
 ```
 
@@ -318,8 +359,9 @@ A `Bucket` trait abstracts this store (local dir always; S3/GCS behind
 `--features s3/gcs`, with conditional PUT for write-once and the lease). The
 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 the complete published model without
-bucket write access or a raw-history mirror. Local commands can inspect and
+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.
 Write-once stores are the collaboration primitive: a viewer encodes and
@@ -363,9 +405,10 @@ follows `Chart.appVersion`, runs each container as UID 10001 with a read-only
 root filesystem, and exposes MCP only as a cluster Service. Remote MCP is
 disabled by default; enabling it requires an application TLS Secret and a
 NetworkPolicy with explicit source selectors and object-store destination
-CIDRs. Its egress is limited to cluster DNS and those ranges on TCP 443; the
-chart accepts at most 64 ranges and rejects IPv4 prefixes broader than `/12`
-and IPv6 prefixes broader than `/32`.
+CIDRs. Athena mode additionally requires explicit AWS API destination CIDRs.
+Its egress is limited to cluster DNS and those ranges on TCP 443; each list is
+capped at 64 entries and rejects IPv4 prefixes broader than `/12` and IPv6
+prefixes broader than `/32`.
 
 ## Data compatibility
 
@@ -382,8 +425,8 @@ and IPv6 prefixes broader than `/32`.
 - **The read-model pointer carries `format` + the writer's version.** A reader
   meeting a newer format refuses to pull it and says to upgrade; an unreadable
   derived blob is a cache miss, never an error. Format 2 publishes compact
-  analysis and trace projections alongside documents, so mediated readers need
-  no raw bucket objects.
+  analysis and local trace projections alongside documents. Athena-backed MCP
+  omits the trace blob and queries bounded raw rows without mirroring them.
 
 ## What's built (kernel)
 
@@ -397,7 +440,8 @@ and the completed pointer local), `search [--filter col=value] [--json]`, `topic
 `mcp` (stdio + optional HTTP), `import`, `cluster [--resolution]`, `summarize`,
 `eval`, plus the scenario test suite (`cargo test`, pure). The bucket backplane
 (local always, S3/GCS opt-in) gives fleet-wide encode-once and collaborative
-builds. MCP-HTTP is an optional mediated transport for remote clients.
+builds. MCP-HTTP is an optional mediated transport for remote clients; the
+`athena` feature adds the read-only S3 trace path.
 Validated at M0/M1 on real data (3,938 docs / 770 K embeddings): retrieval 12/12
 relevant top-3, agent task-start dogfood 3/3, session summaries specific and
 accurate (extractive in the core; one-line abstractive from a local Qwen3-0.6B
diff --git a/src/main.rs b/src/main.rs
index cdddc66..2a3724b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -50,6 +50,8 @@ mod search;
 mod summarize;
 mod topics;
 mod trace;
+#[cfg(feature = "athena")]
+mod trace_athena;
 
 use anyhow::Result;
 use clap::{Parser, Subcommand, ValueEnum};
@@ -440,6 +442,16 @@ enum Cmd {
         /// Response redaction: off, standard, or mcp_safe.
         #[arg(long)]
         redaction: Option<String>,
+        /// Athena workgroup for bounded raw-event trace tools. When omitted,
+        /// trace tools use the local projection.
+        #[arg(long)]
+        athena_workgroup: Option<String>,
+        /// Glue database containing the raw event table.
+        #[arg(long, default_value = "synty")]
+        athena_database: String,
+        /// Glue table whose single `line` column contains event envelopes.
+        #[arg(long, default_value = "raw_events")]
+        athena_table: String,
     },
     /// Session + topic summaries. With the `llm` feature, session summaries are
     /// generated by a local Qwen3 model and cached; topics stay extractive.
@@ -962,11 +974,11 @@ fn main() -> Result<()> {
             role,
             scope,
             redaction,
+            athena_workgroup,
+            athena_database,
+            athena_table,
         } => {
             let bucket = config::resolve_bucket_opt(bucket);
-            if let Some(bucket) = &bucket {
-                sync::pull_read_model_for_read(bucket);
-            }
             policy::validate_scope_path(scope.as_deref())?;
             let scope = policy::ReadScope::load(scope.as_deref())?;
             let role = role.parse()?;
@@ -974,6 +986,17 @@ fn main() -> Result<()> {
                 Some(profile) => profile.parse()?,
                 None => config::mcp_redaction(),
             };
+            let athena = athena_workgroup
+                .or_else(|| std::env::var("SYNTY_ATHENA_WORKGROUP").ok())
+                .filter(|workgroup| !workgroup.is_empty())
+                .map(|workgroup| mcp::AthenaTraceOptions {
+                    workgroup,
+                    database: athena_database,
+                    table: athena_table,
+                });
+            if let Some(bucket) = &bucket {
+                sync::pull_read_model_for_mcp(bucket, athena.is_none());
+            }
             if http {
                 let token = token
                     .or_else(|| std::env::var("SYNTY_MCP_TOKEN").ok())
@@ -990,9 +1013,10 @@ fn main() -> Result<()> {
                     redaction,
                     allowed_origins,
                     bucket,
+                    athena,
                 })?
             } else {
-                mcp::run(model_id(), role, scope, redaction, bucket)?
+                mcp::run(model_id(), role, scope, redaction, bucket, athena)?
             }
         }
         Cmd::Summarize {
@@ -1207,6 +1231,12 @@ mod tests {
             "scope.json",
             "--redaction",
             "mcp_safe",
+            "--athena-workgroup",
+            "synty-mcp-readonly",
+            "--athena-database",
+            "synty",
+            "--athena-table",
+            "raw_events",
             "--allowed-origin",
             "https://memory.example.com",
         ])
@@ -1223,6 +1253,9 @@ mod tests {
                 role,
                 scope: Some(s),
                 redaction: Some(r),
+                athena_workgroup: Some(workgroup),
+                athena_database,
+                athena_table,
                 allowed_origins,
                 ..
             } if b == "s3://team" && bind == "127.0.0.1:9000"
@@ -1232,6 +1265,9 @@ mod tests {
                 && role == "investigator"
                 && s == "scope.json"
                 && r == "mcp_safe"
+                && workgroup == "synty-mcp-readonly"
+                && athena_database == "synty"
+                && athena_table == "raw_events"
                 && allowed_origins == ["https://memory.example.com"]
         ));
     }
diff --git a/src/mcp.rs b/src/mcp.rs
index a2a60e8..e8c6252 100644
--- a/src/mcp.rs
+++ b/src/mcp.rs
@@ -21,9 +21,10 @@ pub fn run(
     scope: crate::policy::ReadScope,
     redaction: crate::redact::Profile,
     bucket: Option<String>,
+    athena: Option<AthenaTraceOptions>,
 ) -> Result<()> {
-    start_bucket_refresh(bucket.clone());
-    let mut srv = Server::new(model_id, role, scope, redaction, true, bucket);
+    start_bucket_refresh(bucket.clone(), athena.is_none());
+    let mut srv = Server::new(model_id, role, scope, redaction, true, bucket, athena);
     eprintln!("synty mcp: serving tools over stdio");
     let stdin = std::io::stdin();
     for line in stdin.lock().lines() {
@@ -41,12 +42,158 @@ pub fn run(
     Ok(())
 }
 
+#[derive(Clone, Debug)]
+#[cfg_attr(not(feature = "athena"), allow(dead_code))]
+pub(crate) struct AthenaTraceOptions {
+    pub(crate) workgroup: String,
+    pub(crate) database: String,
+    pub(crate) table: String,
+}
+
+enum TraceBackend {
+    Local,
+    #[cfg(feature = "athena")]
+    Athena(Box<crate::trace_athena::Backend>),
+    Unavailable(String),
+}
+
+impl TraceBackend {
+    fn new(bucket: Option<&str>, options: Option<AthenaTraceOptions>) -> Self {
+        let Some(options) = options else { return Self::Local };
+        let Some(bucket) = bucket else {
+            return Self::Unavailable("--athena-workgroup requires --bucket s3://...".into());
+        };
+        #[cfg(feature = "athena")]
+        {
+            let result = crate::trace_athena::Config::new(
+                bucket.to_string(),
+                options.workgroup,
+                options.database,
+                options.table,
+            )
+            .and_then(crate::trace_athena::Backend::new);
+            match result {
+                Ok(backend) => Self::Athena(Box::new(backend)),
+                Err(error) => Self::Unavailable(error.to_string()),
+            }
+        }
+        #[cfg(not(feature = "athena"))]
+        {
+            let _ = (bucket, options);
+            Self::Unavailable(
+                "Athena trace is not in this binary; rebuild with --features athena".into(),
+            )
+        }
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn list(
+        &mut self,
+        entity: &str,
+        repo: Option<&str>,
+        machine: Option<&str>,
+        source: Option<&str>,
+        status: Option<&str>,
+        operation: Option<&str>,
+        has_errors: bool,
+        since: Option<&str>,
+        min_ms: Option<u64>,
+        sort: &str,
+        limit: usize,
+        scope: &crate::policy::ReadScope,
+    ) -> Result<String> {
+        match self {
+            Self::Local => trace::list_text(
+                entity,
+                repo,
+                machine,
+                source,
+                status,
+                operation,
+                has_errors,
+                since,
+                min_ms,
+                sort,
+                limit,
+                false,
+                Some(scope),
+            ),
+            #[cfg(feature = "athena")]
+            Self::Athena(backend) => backend.list(
+                entity, repo, machine, source, status, operation, has_errors, since, min_ms, sort,
+                limit, scope,
+            ),
+            Self::Unavailable(error) => Err(anyhow::anyhow!("{error}")),
+        }
+    }
+
+    fn show(
+        &mut self,
+        id: &str,
+        before: usize,
+        after: usize,
+        scope: &crate::policy::ReadScope,
+    ) -> Result<String> {
+        match self {
+            Self::Local => trace::show_text(id, before, after, false, Some(scope)),
+            #[cfg(feature = "athena")]
+            Self::Athena(backend) => backend.show(id, before, after, scope),
+            Self::Unavailable(error) => Err(anyhow::anyhow!("{error}")),
+        }
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn search(
+        &mut self,
+        query: &str,
+        repo: Option<&str>,
+        machine: Option<&str>,
+        source: Option<&str>,
+        kind: Option<&str>,
+        limit: usize,
+        scope: &crate::policy::ReadScope,
+    ) -> Result<String> {
+        match self {
+            Self::Local => trace::search_text(
+                query,
+                repo,
+                machine,
+                source,
+                kind,
+                limit,
+                false,
+                Some(scope),
+            ),
+            #[cfg(feature = "athena")]
+            Self::Athena(backend) => {
+                backend.search(query, repo, machine, source, kind, limit, scope)
+            }
+            Self::Unavailable(error) => Err(anyhow::anyhow!("{error}")),
+        }
+    }
+
+    fn compare(
+        &mut self,
+        left: &str,
+        right: &str,
+        scope: &crate::policy::ReadScope,
+    ) -> Result<String> {
+        match self {
+            Self::Local => trace::compare_text(left, right, false, Some(scope)),
+            #[cfg(feature = "athena")]
+            Self::Athena(backend) => backend.compare(left, right, scope),
+            Self::Unavailable(error) => Err(anyhow::anyhow!("{error}")),
+        }
+    }
+}
+
 pub(crate) struct Server {
     model_id: String,
     role: crate::policy::McpRole,
     scope: crate::policy::ReadScope,
     redaction: crate::redact::Profile,
     allow_repo_paths: bool,
+    trace: TraceBackend,
     /// The build the engine serves + encoder + index — loaded on the first
     /// search call, kept warm, reopened when the pointer moves.
     engine: Option<(readmodel::Current, Encoder, MmapIndex)>,
@@ -59,14 +206,17 @@ impl Server {
         scope: crate::policy::ReadScope,
         redaction: crate::redact::Profile,
         allow_repo_paths: bool,
-        _bucket: Option<String>,
+        bucket: Option<String>,
+        athena: Option<AthenaTraceOptions>,
     ) -> Self {
+        let trace = TraceBackend::new(bucket.as_deref(), athena);
         Self {
             model_id,
             role,
             scope,
             redaction,
             allow_repo_paths,
+            trace,
             engine: None,
         }
     }
@@ -137,7 +287,7 @@ impl Server {
                 if entity.is_empty() {
                     Err(anyhow::anyhow!("type is required (turns, spans, or jobs)"))
                 } else {
-                    trace::list_text(
+                    self.trace.list(
                         entity,
                         string_arg(a, "repo"),
                         string_arg(a, "machine"),
@@ -149,8 +299,7 @@ impl Server {
                         a["min_ms"].as_u64(),
                         a["sort"].as_str().unwrap_or("recent"),
                         bounded_positive(a, "limit", 20, 100),
-                        false,
-                        Some(&self.scope),
+                        &self.scope,
                     )
                 }
             }
@@ -159,12 +308,11 @@ impl Server {
                 if id.is_empty() {
                     Err(anyhow::anyhow!("id is required — trace ids appear in synty_trace_list"))
                 } else {
-                    trace::show_text(
+                    self.trace.show(
                         id,
                         bounded(a, "before", 6, 100),
                         bounded(a, "after", 12, 100),
-                        false,
-                        Some(&self.scope),
+                        &self.scope,
                     )
                 }
             }
@@ -175,15 +323,14 @@ impl Server {
                 } else if query.chars().count() > 4096 {
                     Err(anyhow::anyhow!("query exceeds 4096 characters"))
                 } else {
-                    trace::search_text(
+                    self.trace.search(
                         query,
                         string_arg(a, "repo"),
                         string_arg(a, "machine"),
                         string_arg(a, "source"),
                         string_arg(a, "kind"),
                         bounded_positive(a, "limit", 20, 100),
-                        false,
-                        Some(&self.scope),
+                        &self.scope,
                     )
                 }
             }
@@ -193,7 +340,7 @@ impl Server {
                 if left.is_empty() || right.is_empty() {
                     Err(anyhow::anyhow!("left and right trace ids are required"))
                 } else {
-                    trace::compare_text(left, right, false, Some(&self.scope))
+                    self.trace.compare(left, right, &self.scope)
                 }
             }
             other => return Err(format!("unknown tool: {other}")),
@@ -275,13 +422,12 @@ impl Server {
     }
 }
 
-/// Keep the complete published read model fresh without holding the MCP
-/// dispatcher lock. Format-2 builds include compact session and trace
-/// projections, so a mediated reader never downloads the raw event lake.
-pub(crate) fn start_bucket_refresh(bucket: Option<String>) {
+/// Keep the selected published model fresh without holding the dispatcher.
+/// Athena readers omit trace.json; neither mode mirrors the raw event lake.
+pub(crate) fn start_bucket_refresh(bucket: Option<String>, include_trace: bool) {
     let Some(bucket) = bucket else { return };
     std::thread::spawn(move || loop {
-        crate::sync::pull_read_model_for_read(&bucket);
+        crate::sync::pull_read_model_for_mcp(&bucket, include_trace);
         std::thread::sleep(std::time::Duration::from_secs(30));
     });
 }
@@ -461,6 +607,7 @@ mod tests {
             crate::redact::Profile::Off,
             true,
             None,
+            None,
         )
     }
 
@@ -494,6 +641,7 @@ mod tests {
             crate::redact::Profile::Off,
             true,
             None,
+            None,
         );
         let listed = server.handle(&json!({"jsonrpc":"2.0","id":1,"method":"tools/list"})).unwrap();
         let names: Vec<&str> = listed["result"]["tools"].as_array().unwrap().iter()
diff --git a/src/mcp_http.rs b/src/mcp_http.rs
index ea1bcd2..f756d2f 100644
--- a/src/mcp_http.rs
+++ b/src/mcp_http.rs
@@ -73,6 +73,17 @@ struct DispatchJob {
 #[derive(Clone)]
 struct Dispatcher {
     sender: SyncSender<DispatchJob>,
+    alive: Arc<AtomicBool>,
+}
+
+#[cfg(feature = "mcp-http")]
+struct DispatcherLife(Arc<AtomicBool>);
+
+#[cfg(feature = "mcp-http")]
+impl Drop for DispatcherLife {
+    fn drop(&mut self) {
+        self.0.store(false, Ordering::Release);
+    }
 }
 
 #[cfg(feature = "mcp-http")]
@@ -85,6 +96,7 @@ struct AppState {
     rate_limiter: Arc<Mutex<RateLimiter>>,
     in_flight: Arc<Semaphore>,
     require_read_model: bool,
+    require_trace_projection: bool,
 }
 
 #[cfg(feature = "mcp-http")]
@@ -107,9 +119,12 @@ enum DispatchError {
 impl Dispatcher {
     fn start(name: &str, queue: usize, mut server: mcp::Server) -> Result<Self> {
         let (sender, receiver) = mpsc::sync_channel::<DispatchJob>(queue);
+        let alive = Arc::new(AtomicBool::new(true));
+        let thread_alive = Arc::clone(&alive);
         std::thread::Builder::new()
             .name(format!("synty-mcp-{name}"))
             .spawn(move || {
+                let _life = DispatcherLife(thread_alive);
                 while let Ok(job) = receiver.recv() {
                     if job.cancelled.load(Ordering::Acquire) {
                         continue;
@@ -119,7 +134,7 @@ impl Dispatcher {
                 }
             })
             .map_err(|error| anyhow::anyhow!("start {name} MCP dispatcher: {error}"))?;
-        Ok(Self { sender })
+        Ok(Self { sender, alive })
     }
 
     fn call(&self, request: Value) -> std::result::Result<Option<Value>, DispatchError> {
@@ -140,6 +155,10 @@ impl Dispatcher {
             Err(RecvTimeoutError::Disconnected) => Err(DispatchError::Stopped),
         }
     }
+
+    fn is_alive(&self) -> bool {
+        self.alive.load(Ordering::Acquire)
+    }
 }
 
 #[cfg(any(feature = "mcp-http", test))]
@@ -187,6 +206,7 @@ pub struct Opts {
     pub redaction: crate::redact::Profile,
     pub allowed_origins: Vec<String>,
     pub bucket: Option<String>,
+    pub athena: Option<crate::mcp::AthenaTraceOptions>,
 }
 
 #[cfg(feature = "mcp-http")]
@@ -196,7 +216,8 @@ pub fn run(opts: Opts) -> Result<()> {
     validate_listener(&opts.bind, opts.listen_public, tls.is_some())?;
     let scheme = if tls.is_some() { "https" } else { "http" };
     let require_read_model = opts.bucket.is_some();
-    mcp::start_bucket_refresh(opts.bucket.clone());
+    let require_trace_projection = opts.athena.is_none();
+    mcp::start_bucket_refresh(opts.bucket.clone(), require_trace_projection);
     let dispatcher = Dispatcher::start("semantic", DISPATCH_QUEUE, mcp::Server::new(
         opts.model_id.clone(),
         opts.role,
@@ -204,12 +225,13 @@ pub fn run(opts: Opts) -> Result<()> {
         opts.redaction,
         false,
         opts.bucket.clone(),
+        opts.athena.clone(),
     ))?;
     // Analysis tools share a cached published projection. Keep them serialized
     // on a one-slot queue so bursts cannot duplicate its first-load memory or
     // block semantic search.
     let analysis_dispatcher = Dispatcher::start("analysis", ANALYSIS_DISPATCH_QUEUE, mcp::Server::new(
-        opts.model_id, opts.role, opts.scope, opts.redaction, false, opts.bucket,
+        opts.model_id, opts.role, opts.scope, opts.redaction, false, opts.bucket, opts.athena,
     ))?;
     let state = AppState {
         dispatcher,
@@ -219,6 +241,7 @@ pub fn run(opts: Opts) -> Result<()> {
         rate_limiter: Arc::new(Mutex::new(RateLimiter::default())),
         in_flight: Arc::new(Semaphore::new(MAX_IN_FLIGHT)),
         require_read_model,
+        require_trace_projection,
     };
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
@@ -364,14 +387,23 @@ async fn handle(
     }
     if request.method() == Method::GET && matches!(path, "/health" | "/ready") {
         let read_model = crate::readmodel::current().is_some();
-        let mediated = crate::readmodel::mediated_ready();
-        let ready = !state.require_read_model || (read_model && mediated);
+        let mediated = crate::readmodel::mediated_ready(state.require_trace_projection);
+        let dispatchers_ready =
+            state.dispatcher.is_alive() && state.analysis_dispatcher.is_alive();
+        let ready = ready_state(
+            state.require_read_model,
+            read_model,
+            mediated,
+            dispatchers_ready,
+        );
         let liveness = path == "/health";
         let body = json!({
             "status": if liveness || ready { "ok" } else { "starting" },
             "ready": ready,
             "read_model": read_model,
             "mediated_projections": mediated,
+            "dispatchers": dispatchers_ready,
+            "trace_backend": if state.require_trace_projection { "projection" } else { "athena" },
             "name": "synty",
             "version": env!("CARGO_PKG_VERSION"),
         })
@@ -503,6 +535,16 @@ fn health_code(ready: bool) -> u16 {
     if ready { 200 } else { 503 }
 }
 
+#[cfg(any(feature = "mcp-http", test))]
+fn ready_state(
+    require_read_model: bool,
+    read_model: bool,
+    mediated: bool,
+    dispatchers: bool,
+) -> bool {
+    dispatchers && (!require_read_model || (read_model && mediated))
+}
+
 #[cfg(any(feature = "mcp-http", test))]
 fn uses_analysis_projection(request: &serde_json::Value) -> bool {
     request["method"].as_str() == Some("tools/call")
@@ -645,6 +687,9 @@ 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));
     }
 
     #[test]
@@ -721,6 +766,7 @@ mod tests {
                 crate::redact::Profile::Off,
                 false,
                 None,
+                None,
             ),
         )
         .unwrap();
diff --git a/src/readmodel.rs b/src/readmodel.rs
index f326ac5..44c3011 100644
--- a/src/readmodel.rs
+++ b/src/readmodel.rs
@@ -120,13 +120,15 @@ pub fn trace_path() -> PathBuf {
         .unwrap_or_else(|| PathBuf::from("corpus/trace.json"))
 }
 
-/// Strict readiness for mediated readers: format-2 builds carry the compact
-/// raw-derived projections that keep requests bounded independently of raw
-/// retention. Older search-only builds remain usable by local CLI fallbacks.
+/// Strict readiness for mediated readers. Athena-backed trace readers require
+/// only the compact analysis projection; local trace readers additionally
+/// require the published trace snapshot.
 #[cfg(feature = "mcp-http")]
-pub fn mediated_ready() -> bool {
+pub fn mediated_ready(require_trace: bool) -> bool {
     current().is_some_and(|current| {
-        current.format >= FORMAT && current.analysis().is_file() && current.trace().is_file()
+        current.format >= FORMAT
+            && current.analysis().is_file()
+            && (!require_trace || current.trace().is_file())
     })
 }
 
diff --git a/src/sync.rs b/src/sync.rs
index 673f340..6cd6823 100644
--- a/src/sync.rs
+++ b/src/sync.rs
@@ -26,6 +26,17 @@ struct BuildManifest {
     files: BTreeMap<String, String>,
 }
 
+fn selected_read_model_files(
+    files: &BTreeMap<String, String>,
+    include_trace: bool,
+) -> BTreeMap<String, String> {
+    files
+        .iter()
+        .filter(|(name, _)| include_trace || name.as_str() != "trace.json")
+        .map(|(name, blob)| (name.clone(), blob.clone()))
+        .collect()
+}
+
 fn manifest_key(build: &str, rev: u64) -> String {
     format!("builds/{build}.{rev}.json")
 }
@@ -693,10 +704,16 @@ enum ReaderPull {
 // complete format-2 read model.
 const READER_PULL_ORDER: [ReaderPull; 2] = [ReaderPull::ReadModel, ReaderPull::Events];
 
-/// Pull only the complete published query model. Format-2 builds include the
-/// compact analysis and trace projections needed by mediated MCP readers.
+/// Pull the complete published query model for local readers.
 pub fn pull_read_model_for_read(bucket_uri: &str) {
-    match pull_if_stale(bucket_uri) {
+    pull_read_model_for_mcp(bucket_uri, true);
+}
+
+/// Pull the published model for a mediated reader. Athena-backed trace readers
+/// deliberately omit the legacy trace blob; semantic search and compact
+/// analysis remain local and immutable.
+pub fn pull_read_model_for_mcp(bucket_uri: &str, include_trace: bool) {
+    match pull_if_stale_selected(bucket_uri, include_trace) {
         Ok(true) => eprintln!("pulled published read-model from {bucket_uri}"),
         Ok(false) => {}
         Err(e) => eprintln!("read-model pull skipped ({e})"),
@@ -862,6 +879,10 @@ pub fn publish(bucket_uri: &str) -> Result<usize> {
 /// verified loadable, and only then does the local pointer move. Returns
 /// whether it pulled.
 pub fn pull_if_stale(bucket_uri: &str) -> Result<bool> {
+    pull_if_stale_selected(bucket_uri, true)
+}
+
+fn pull_if_stale_selected(bucket_uri: &str, include_trace: bool) -> Result<bool> {
     let b = bucket::open(bucket_uri)?;
     let Some(remote) =
         b.get(POINTER_KEY)?.and_then(|raw| serde_json::from_slice::<readmodel::Current>(&raw).ok())
@@ -879,18 +900,20 @@ pub fn pull_if_stale(bucket_uri: &str) -> Result<bool> {
     if local_format_is_newer(local.as_ref(), &remote) {
         return Ok(false);
     }
-    if local.as_ref() == Some(&remote) {
-        return Ok(false);
-    }
-
     let raw = b
         .get(&manifest_key(&remote.build, remote.rev))?
         .ok_or_else(|| anyhow!("bucket pointer names a missing manifest (publish in flight?)"))?;
     let manifest: BuildManifest = serde_json::from_slice(&raw)?;
+    let files = selected_read_model_files(&manifest.files, include_trace);
     let dir = readmodel::build_dir(&remote.build);
+    if local.as_ref() == Some(&remote)
+        && files.keys().all(|name| dir.join(name).is_file())
+    {
+        return Ok(false);
+    }
     // Transfer the build — hardlink-reuse on-disk blobs, fetch only the changed
     // ones (and record the local manifest for the next pull).
-    let (reused, fetched, bytes_down) = fetch_build(&*b, &dir, &manifest.files)?;
+    let (reused, fetched, bytes_down) = fetch_build(&*b, &dir, &files)?;
     // Verify before repointing — a bad pull must never become current.
     next_plaid::MmapIndex::load(&dir.to_string_lossy())
         .map_err(|e| anyhow!("pulled build does not load: {e}"))?;
@@ -920,6 +943,20 @@ mod tests {
         assert_eq!(READER_PULL_ORDER, [ReaderPull::ReadModel, ReaderPull::Events]);
     }
 
+    #[test]
+    fn athena_mcp_pull_omits_only_the_legacy_trace_blob() {
+        let files = BTreeMap::from([
+            ("analysis.json".into(), "analysis".into()),
+            ("docs.jsonl".into(), "docs".into()),
+            ("index.db".into(), "index".into()),
+            ("trace.json".into(), "trace".into()),
+        ]);
+        let athena = selected_read_model_files(&files, false);
+        assert!(!athena.contains_key("trace.json"));
+        assert_eq!(athena.len(), files.len() - 1);
+        assert_eq!(selected_read_model_files(&files, true), files);
+    }
+
     #[test]
     fn older_remote_format_never_replaces_a_newer_local_projection() {
         let local = readmodel::Current {
diff --git a/src/trace.rs b/src/trace.rs
index 3e9afb8..cdcde07 100644
--- a/src/trace.rs
+++ b/src/trace.rs
@@ -149,7 +149,7 @@ struct TurnBuilder {
 }
 
 #[derive(Default)]
-struct TraceStore {
+pub(crate) struct TraceStore {
     events: Vec<TraceEvent>,
     spans: Vec<Span>,
     jobs: Vec<Job>,
@@ -446,6 +446,17 @@ impl TraceStore {
         Ok(builder.finish())
     }
 
+    /// Fold a bounded raw-event selection into the same query model used by
+    /// local projections. Remote readers use this after Athena has already
+    /// pruned the event lake by stream and time.
+    #[cfg(feature = "athena")]
+    pub(crate) fn from_text(text: &str) -> Self {
+        let known: HashSet<String> = crate::config::load().repos.into_iter().collect();
+        let mut builder = SnapshotBuilder::new(known, None);
+        builder.fold_text(text);
+        builder.finish()
+    }
+
     #[cfg(test)]
     fn from_lines(lines: &[&str]) -> Self {
         let mut builder = SnapshotBuilder::new(HashSet::new(), None);
@@ -1029,6 +1040,29 @@ pub fn list_text(
     scope: Option<&crate::policy::ReadScope>,
 ) -> Result<String> {
     let store = TraceStore::load()?;
+    list_store_text(
+        &store, entity, repo, machine, source, status, operation, has_errors, since, min_ms, sort,
+        limit, json_out, scope,
+    )
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn list_store_text(
+    store: &TraceStore,
+    entity: &str,
+    repo: Option<&str>,
+    machine: Option<&str>,
+    source: Option<&str>,
+    status: Option<&str>,
+    operation: Option<&str>,
+    has_errors: bool,
+    since: Option<&str>,
+    min_ms: Option<u64>,
+    sort: &str,
+    limit: usize,
+    json_out: bool,
+    scope: Option<&crate::policy::ReadScope>,
+) -> Result<String> {
     let out = match entity {
         "turns" => {
             let mut rows: Vec<&Turn> = store
@@ -1051,7 +1085,7 @@ pub fn list_text(
                     )
                 })
                 .filter(|t| !has_errors || t.errors > 0)
-                .filter(|t| scope_allows(&store, &t.session_id, &t.repo, &t.source, scope))
+                .filter(|t| scope_allows(store, &t.session_id, &t.repo, &t.source, scope))
                 .filter(|t| {
                     operation.is_none_or(|q| {
                         t.span_indexes
@@ -1089,7 +1123,7 @@ pub fn list_text(
                     )
                 })
                 .filter(|s| operation.is_none_or(|q| span_operation_matches(s, q)))
-                .filter(|s| scope_allows(&store, &s.session_id, &s.repo, &s.source, scope))
+                .filter(|s| scope_allows(store, &s.session_id, &s.repo, &s.source, scope))
                 .collect();
             sort_spans(&mut rows, sort);
             rows.truncate(limit);
@@ -1120,7 +1154,7 @@ pub fn list_text(
                     )
                 })
                 .filter(|j| !has_errors || j.errors > 0)
-                .filter(|j| scope_allows(&store, &j.session_id, &j.repo, &j.source, scope))
+                .filter(|j| scope_allows(store, &j.session_id, &j.repo, &j.source, scope))
                 .filter(|j| operation.is_none_or(|q| job_operation_matches(j, q)))
                 .collect();
             sort_jobs(&mut rows, sort);
@@ -1248,14 +1282,25 @@ pub fn show_text(
     scope: Option<&crate::policy::ReadScope>,
 ) -> Result<String> {
     let store = TraceStore::load()?;
-    let resolved = resolve(&store, id, scope)?;
-    anyhow::ensure!(resolved_allowed(&store, &resolved, scope), "trace id is outside the read scope");
+    show_store_text(&store, id, before, after, json_out, scope)
+}
+
+pub(crate) fn show_store_text(
+    store: &TraceStore,
+    id: &str,
+    before: usize,
+    after: usize,
+    json_out: bool,
+    scope: Option<&crate::policy::ReadScope>,
+) -> Result<String> {
+    let resolved = resolve(store, id, scope)?;
+    anyhow::ensure!(resolved_allowed(store, &resolved, scope), "trace id is outside the read scope");
     match resolved {
-        Resolved::Job(i) => show_job_text(&store, i, json_out),
-        Resolved::Turn(i) => show_turn_text(&store, i, json_out),
-        Resolved::Span(i) => show_span_text(&store, i, before, after, json_out),
-        Resolved::Event(i) => show_event_text(&store, i, before, after, json_out),
-        Resolved::Session(sid) => show_session_text(&store, &sid, json_out),
+        Resolved::Job(i) => show_job_text(store, i, json_out),
+        Resolved::Turn(i) => show_turn_text(store, i, json_out),
+        Resolved::Span(i) => show_span_text(store, i, before, after, json_out),
+        Resolved::Event(i) => show_event_text(store, i, before, after, json_out),
+        Resolved::Session(sid) => show_session_text(store, &sid, json_out),
     }
 }
 
@@ -1271,15 +1316,25 @@ pub fn compare_text(
     scope: Option<&crate::policy::ReadScope>,
 ) -> Result<String> {
     let store = TraceStore::load()?;
-    let left_resolved = resolve(&store, left, scope)?;
-    let right_resolved = resolve(&store, right, scope)?;
+    compare_store_text(&store, left, right, json_out, scope)
+}
+
+pub(crate) fn compare_store_text(
+    store: &TraceStore,
+    left: &str,
+    right: &str,
+    json_out: bool,
+    scope: Option<&crate::policy::ReadScope>,
+) -> Result<String> {
+    let left_resolved = resolve(store, left, scope)?;
+    let right_resolved = resolve(store, right, scope)?;
     anyhow::ensure!(
-        resolved_allowed(&store, &left_resolved, scope)
-            && resolved_allowed(&store, &right_resolved, scope),
+        resolved_allowed(store, &left_resolved, scope)
+            && resolved_allowed(store, &right_resolved, scope),
         "trace id is outside the read scope"
     );
-    let l = comparable(&store, left_resolved)?;
-    let r = comparable(&store, right_resolved)?;
+    let l = comparable(store, left_resolved)?;
+    let r = comparable(store, right_resolved)?;
     let mut differences = Vec::new();
     if let (Some(lm), Some(rm)) = (l.as_object(), r.as_object()) {
         let mut keys: Vec<&String> = lm.keys().chain(rm.keys()).collect();
@@ -1327,7 +1382,25 @@ pub fn search_text(
         bail!("trace search: query cannot be empty");
     }
     let store = TraceStore::load()?;
-    let hits = search_hits(&store, query, repo, machine, source, kind, limit, scope);
+    search_store_text(&store, query, repo, machine, source, kind, limit, json_out, scope)
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn search_store_text(
+    store: &TraceStore,
+    query: &str,
+    repo: Option<&str>,
+    machine: Option<&str>,
+    source: Option<&str>,
+    kind: Option<&str>,
+    limit: usize,
+    json_out: bool,
+    scope: Option<&crate::policy::ReadScope>,
+) -> Result<String> {
+    if query.trim().is_empty() {
+        bail!("trace search: query cannot be empty");
+    }
+    let hits = search_hits(store, query, repo, machine, source, kind, limit, scope);
     if json_out {
         Ok(crate::view::envelope("trace_search", json!(hits)))
     } else {
diff --git a/src/trace_athena.rs b/src/trace_athena.rs
new file mode 100644
index 0000000..3482705
--- /dev/null
+++ b/src/trace_athena.rs
@@ -0,0 +1,911 @@
+// Read-only remote trace access over Athena. SQL prunes the immutable event
+// 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 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::time::{Duration as StdDuration, Instant};
+
+const DEFAULT_LIST_HOURS: i64 = 1;
+const DEFAULT_LOOKUP_HOURS: i64 = 24 * 7;
+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 AWS_CONNECT_TIMEOUT: StdDuration = StdDuration::from_secs(3);
+const AWS_OPERATION_TIMEOUT: StdDuration = StdDuration::from_secs(10);
+
+#[derive(Clone, Debug)]
+pub(crate) struct Config {
+    pub bucket: String,
+    pub workgroup: String,
+    pub database: String,
+    pub table: String,
+}
+
+impl Config {
+    pub(crate) fn new(
+        bucket: String,
+        workgroup: String,
+        database: String,
+        table: String,
+    ) -> Result<Self> {
+        anyhow::ensure!(
+            bucket.starts_with("s3://"),
+            "Athena trace requires an s3:// bucket"
+        );
+        validate_name("Athena workgroup", &workgroup, false)?;
+        validate_name("Glue database", &database, true)?;
+        validate_name("Glue table", &table, true)?;
+        Ok(Self {
+            bucket,
+            workgroup,
+            database,
+            table,
+        })
+    }
+}
+
+struct QueryRows {
+    lines: Vec<String>,
+}
+
+trait EventQuery: Send {
+    fn run(&mut self, sql: &str) -> Result<QueryRows>;
+}
+
+struct AwsEventQuery {
+    client: aws_sdk_athena::Client,
+    runtime: tokio::runtime::Runtime,
+    workgroup: String,
+    database: String,
+}
+
+impl AwsEventQuery {
+    fn new(config: &Config) -> Result<Self> {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .context("start Athena runtime")?;
+        let profile = crate::config::load().aws_profile;
+        let timeout = TimeoutConfig::builder()
+            .connect_timeout(AWS_CONNECT_TIMEOUT)
+            .operation_attempt_timeout(AWS_OPERATION_TIMEOUT)
+            .operation_timeout(AWS_OPERATION_TIMEOUT)
+            .build();
+        let mut loader =
+            aws_config::defaults(aws_config::BehaviorVersion::latest()).timeout_config(timeout);
+        if let Some(profile) = profile {
+            loader = loader.profile_name(profile);
+        }
+        let sdk = runtime.block_on(loader.load());
+        Ok(Self {
+            client: aws_sdk_athena::Client::new(&sdk),
+            runtime,
+            workgroup: config.workgroup.clone(),
+            database: config.database.clone(),
+        })
+    }
+}
+
+impl EventQuery for AwsEventQuery {
+    fn run(&mut self, sql: &str) -> Result<QueryRows> {
+        anyhow::ensure!(
+            sql.trim_start().to_ascii_uppercase().starts_with("SELECT "),
+            "Athena trace only permits SELECT statements"
+        );
+        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 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) => {
+                    break 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")
+                    );
+                }
+                _ => 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 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;
+            }
+        }
+        metrics::Run::new("athena_trace")
+            .set("rows", lines.len())
+            .set("result_bytes", bytes)
+            .set("scanned_bytes", scanned_bytes)
+            .set("elapsed_ms", started.elapsed().as_millis() as u64)
+            .emit();
+        Ok(QueryRows { lines })
+    }
+}
+
+pub(crate) struct Backend {
+    config: Config,
+    query: Box<dyn EventQuery>,
+    // 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<Vec<String>>,
+    cached: Option<trace::TraceStore>,
+}
+
+impl Backend {
+    pub(crate) fn new(config: Config) -> Result<Self> {
+        let query = Box::new(AwsEventQuery::new(&config)?);
+        Ok(Self {
+            config,
+            query,
+            streams: None,
+            cached: None,
+        })
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn list(
+        &mut self,
+        entity: &str,
+        repo: Option<&str>,
+        machine: Option<&str>,
+        source: Option<&str>,
+        status: Option<&str>,
+        operation: Option<&str>,
+        has_errors: bool,
+        since: Option<&str>,
+        min_ms: Option<u64>,
+        sort: &str,
+        limit: usize,
+        scope: &ReadScope,
+    ) -> Result<String> {
+        let window = Window::parse(since, None, DEFAULT_LIST_HOURS)?;
+        let store = self.load_store(
+            window,
+            machine,
+            source,
+            operation,
+            None,
+            &[],
+            operation.is_some(),
+            scope,
+        )?;
+        let out = trace::list_store_text(
+            &store,
+            entity,
+            repo,
+            machine,
+            source,
+            status,
+            operation,
+            has_errors,
+            since,
+            min_ms,
+            sort,
+            limit,
+            false,
+            Some(scope),
+        )?;
+        Ok(out)
+    }
+
+    pub(crate) fn show(
+        &mut self,
+        id: &str,
+        before: usize,
+        after: usize,
+        scope: &ReadScope,
+    ) -> Result<String> {
+        if let Some(store) = &self.cached
+            && let Ok(out) = trace::show_store_text(store, id, before, after, false, Some(scope))
+        {
+            return Ok(out);
+        }
+        let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?;
+        let store = self.load_store(window, None, None, None, None, &[id], true, scope)?;
+        let out = trace::show_store_text(&store, id, before, after, false, Some(scope))?;
+        self.cached = Some(store);
+        Ok(out)
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn search(
+        &mut self,
+        query: &str,
+        repo: Option<&str>,
+        machine: Option<&str>,
+        source: Option<&str>,
+        kind: Option<&str>,
+        limit: usize,
+        scope: &ReadScope,
+    ) -> Result<String> {
+        let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?;
+        let store = self.load_store(
+            window,
+            machine,
+            source,
+            Some(query),
+            kind,
+            &[],
+            false,
+            scope,
+        )?;
+        let out = trace::search_store_text(
+            &store,
+            query,
+            repo,
+            machine,
+            source,
+            kind,
+            limit,
+            false,
+            Some(scope),
+        )?;
+        Ok(out)
+    }
+
+    pub(crate) fn compare(&mut self, left: &str, right: &str, scope: &ReadScope) -> Result<String> {
+        if let Some(store) = &self.cached
+            && let Ok(out) = trace::compare_store_text(store, left, right, false, Some(scope))
+        {
+            return Ok(out);
+        }
+        let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?;
+        let store = self.load_store(window, None, None, None, None, &[left, right], true, scope)?;
+        let out = trace::compare_store_text(&store, left, right, false, Some(scope))?;
+        self.cached = Some(store);
+        Ok(out)
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn load_store(
+        &mut self,
+        window: Window,
+        machine: Option<&str>,
+        source: Option<&str>,
+        needle: Option<&str>,
+        kind: Option<&str>,
+        ids: &[&str],
+        expand_matching_sessions: bool,
+        scope: &ReadScope,
+    ) -> Result<trace::TraceStore> {
+        let streams = self.selected_streams(machine)?;
+        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(),
+            ..Default::default()
+        };
+        let first = self.select(&streams, window, &predicate)?;
+        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"
+        );
+        let context_window = Window {
+            since: std::cmp::max(
+                window.since - Duration::days(1),
+                window.until - Duration::hours(MAX_LOOKBACK_HOURS),
+            ),
+            until: window.until,
+        };
+        let expands_sessions = (!ids.is_empty() || (needle.is_some() && expand_matching_sessions))
+            && !sessions.is_empty();
+        let mut lines = if expands_sessions {
+            self.select(
+                &streams,
+                context_window,
+                &Predicate {
+                    sessions: sessions.iter().cloned().collect(),
+                    scope_sources: scope.sources.clone(),
+                    ..Default::default()
+                },
+            )?
+            .lines
+        } else {
+            first.lines
+        };
+        if !sessions.is_empty() && !expands_sessions {
+            let mut contexts = self
+                .select(
+                    &streams,
+                    context_window,
+                    &Predicate {
+                        sessions: sessions.into_iter().collect(),
+                        kinds: vec!["session_start".into(), "agent_meta".into()],
+                        scope_sources: scope.sources.clone(),
+                        dedupe_session_kinds: true,
+                        ..Default::default()
+                    },
+                )?
+                .lines;
+            contexts.append(&mut lines);
+            lines = contexts;
+        }
+        Ok(trace::TraceStore::from_text(&lines.join("\n")))
+    }
+
+    fn select(
+        &mut self,
+        streams: &[String],
+        window: Window,
+        predicate: &Predicate,
+    ) -> Result<QueryRows> {
+        let sql = select_sql(&self.config, streams, window, predicate, MAX_EVENTS + 1)?;
+        self.query.run(&sql)
+    }
+
+    fn selected_streams(&mut self, machine: Option<&str>) -> Result<Vec<String>> {
+        let mut streams = if let Some(streams) = &self.streams {
+            streams.clone()
+        } else {
+            let keys = bucket::open(&self.config.bucket)?.list("event-streams/")?;
+            let streams: Vec<String> = keys
+                .into_iter()
+                .filter_map(|key| key.strip_prefix("event-streams/").map(str::to_string))
+                .filter(|stream| !stream.is_empty() && !stream.contains('/'))
+                .collect();
+            anyhow::ensure!(
+                !streams.is_empty(),
+                "no event streams found in {}",
+                self.config.bucket
+            );
+            streams
+        };
+        if let Some(machine) = machine {
+            let machine = machine.to_ascii_lowercase();
+            streams.retain(|stream| stream.to_ascii_lowercase().contains(&machine));
+        }
+        anyhow::ensure!(
+            !streams.is_empty(),
+            "no event streams match the machine filter"
+        );
+        Ok(streams)
+    }
+}
+
+#[derive(Default)]
+struct Predicate {
+    source: Option<String>,
+    scope_sources: Vec<String>,
+    needle: Option<String>,
+    kind_contains: Option<String>,
+    ids: Vec<String>,
+    sessions: Vec<String>,
+    kinds: Vec<String>,
+    dedupe_session_kinds: bool,
+}
+
+#[derive(Clone, Copy)]
+struct Window {
+    since: DateTime<Utc>,
+    until: DateTime<Utc>,
+}
+
+impl Window {
+    fn parse(since: Option<&str>, until: Option<&str>, default_hours: i64) -> Result<Self> {
+        let until = match until {
+            Some(value) => parse_time(value)?,
+            None => Utc::now(),
+        };
+        let since = match since {
+            Some(value) => parse_time(value)?,
+            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"
+        );
+        Ok(Self { since, until })
+    }
+}
+
+fn parse_time(value: &str) -> Result<DateTime<Utc>> {
+    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
+        return Ok(timestamp.with_timezone(&Utc));
+    }
+    let date = NaiveDate::parse_from_str(value, "%Y-%m-%d")
+        .with_context(|| format!("trace time must be RFC3339 or YYYY-MM-DD: {value}"))?;
+    Ok(date
+        .and_hms_opt(0, 0, 0)
+        .expect("midnight exists")
+        .and_utc())
+}
+
+fn select_sql(
+    config: &Config,
+    streams: &[String],
+    window: Window,
+    predicate: &Predicate,
+    limit: usize,
+) -> Result<String> {
+    anyhow::ensure!(
+        !streams.is_empty(),
+        "Athena trace needs at least one stream"
+    );
+    let stream_values = streams
+        .iter()
+        .map(|stream| sql_string(stream))
+        .collect::<Vec<_>>()
+        .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!(
+            "from_iso8601_timestamp(json_extract_scalar(line, '$.ts')) >= from_iso8601_timestamp({})",
+            sql_string(&window.since.to_rfc3339())
+        ),
+        format!(
+            "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",
+            sql_string(&source.to_ascii_lowercase())
+        ));
+    }
+    if !predicate.scope_sources.is_empty() {
+        clauses.push(format!(
+            "json_extract_scalar(line, '$.source') IN ({})",
+            predicate
+                .scope_sources
+                .iter()
+                .map(|source| sql_string(source))
+                .collect::<Vec<_>>()
+                .join(", ")
+        ));
+    }
+    if let Some(needle) = predicate.needle.as_deref() {
+        clauses.push(format!(
+            "strpos(lower(line), {}) > 0",
+            sql_string(&needle.to_ascii_lowercase())
+        ));
+    }
+    if let Some(kind) = predicate.kind_contains.as_deref() {
+        clauses.push(format!(
+            "strpos(lower(coalesce(json_extract_scalar(line, '$.kind'), '')), {}) > 0",
+            sql_string(&kind.to_ascii_lowercase())
+        ));
+    }
+    if !predicate.ids.is_empty() {
+        let matches = predicate
+            .ids
+            .iter()
+            .map(|id| {
+                format!(
+                    "(starts_with(coalesce(json_extract_scalar(line, '$.event_id'), ''), {id}) OR \
+                     starts_with(coalesce(json_extract_scalar(line, '$.session_id'), ''), {id}) OR \
+                     strpos(line, {id}) > 0)",
+                    id = sql_string(id)
+                )
+            })
+            .collect::<Vec<_>>()
+            .join(" OR ");
+        clauses.push(format!("({matches})"));
+    }
+    if !predicate.sessions.is_empty() {
+        clauses.push(format!(
+            "json_extract_scalar(line, '$.session_id') IN ({})",
+            predicate
+                .sessions
+                .iter()
+                .map(|session| sql_string(session))
+                .collect::<Vec<_>>()
+                .join(", ")
+        ));
+    }
+    if !predicate.kinds.is_empty() {
+        clauses.push(format!(
+            "json_extract_scalar(line, '$.kind') IN ({})",
+            predicate
+                .kinds
+                .iter()
+                .map(|kind| sql_string(kind))
+                .collect::<Vec<_>>()
+                .join(", ")
+        ));
+    }
+    let where_clause = clauses.join("\n  AND ");
+    if predicate.dedupe_session_kinds {
+        Ok(format!(
+            "SELECT line FROM (\n  SELECT line, row_number() OVER (\n    \
+             PARTITION BY json_extract_scalar(line, '$.session_id'), \
+             json_extract_scalar(line, '$.kind')\n    \
+             ORDER BY json_extract_scalar(line, '$.ts')\n  ) AS synty_context_rank\n  \
+             FROM \"{}\".\"{}\"\n  WHERE {}\n)\n\
+             WHERE synty_context_rank = 1\n\
+             ORDER BY json_extract_scalar(line, '$.ts')\nLIMIT {limit}",
+            config.database, config.table, where_clause,
+        ))
+    } else {
+        Ok(format!(
+            "SELECT line FROM \"{}\".\"{}\"\nWHERE {}\n\
+             ORDER BY json_extract_scalar(line, '$.ts'), \
+             try_cast(json_extract_scalar(line, '$.seq') AS bigint)\nLIMIT {limit}",
+            config.database, config.table, where_clause,
+        ))
+    }
+}
+
+fn event_sessions(lines: &[String]) -> Result<BTreeSet<String>> {
+    let mut sessions = BTreeSet::new();
+    for line in lines {
+        let event: crate::event::Event =
+            serde_json::from_str(line).context("parse Athena event row")?;
+        if !event.session_id.is_empty() {
+            sessions.insert(event.session_id);
+        }
+    }
+    Ok(sessions)
+}
+
+fn sql_string(value: &str) -> String {
+    format!("'{}'", value.replace('\'', "''"))
+}
+
+fn validate_name(label: &str, value: &str, lowercase_only: bool) -> Result<()> {
+    let valid = !value.is_empty()
+        && value.len() <= 128
+        && value.chars().all(|ch| {
+            ch.is_ascii_lowercase()
+                || ch.is_ascii_digit()
+                || ch == '_'
+                || (!lowercase_only && (ch.is_ascii_uppercase() || matches!(ch, '.' | '-')))
+        });
+    if !valid {
+        bail!("{label} contains unsupported characters");
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+    use std::sync::{Arc, Mutex};
+
+    struct FakeQuery {
+        lines: Vec<String>,
+        calls: Arc<Mutex<Vec<String>>>,
+    }
+
+    impl EventQuery for FakeQuery {
+        fn run(&mut self, sql: &str) -> Result<QueryRows> {
+            self.calls.lock().unwrap().push(sql.to_string());
+            Ok(QueryRows {
+                lines: self.lines.clone(),
+            })
+        }
+    }
+
+    fn event(id: &str, ts: &str, kind: &str, payload: serde_json::Value) -> String {
+        json!({
+            "v": 1, "event_id": id, "stream": "edge-m-codex", "seq": 1,
+            "ts": ts, "source": "codex_cli", "session_id": "session-1",
+            "kind": kind, "payload": payload, "rollup_dim": ""
+        })
+        .to_string()
+    }
+
+    #[test]
+    fn bounded_select_uses_both_physical_partitions_and_never_writes() {
+        let config = Config::new(
+            "s3://bucket".into(),
+            "wg".into(),
+            "synty".into(),
+            "raw_events".into(),
+        )
+        .unwrap();
+        let sql = select_sql(
+            &config,
+            &["edge-m-codex".into()],
+            Window {
+                since: parse_time("2026-07-22T10:00:00Z").unwrap(),
+                until: parse_time("2026-07-22T11:00:00Z").unwrap(),
+            },
+            &Predicate::default(),
+            101,
+        )
+        .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'"));
+        for mutating in ["INSERT", "UPDATE", "DELETE", "CREATE", "UNLOAD", "CTAS"] {
+            assert!(!sql.to_ascii_uppercase().contains(mutating), "{mutating}");
+        }
+    }
+
+    #[test]
+    fn raw_athena_rows_reconstruct_the_existing_span_surface() {
+        let lines = vec![
+            event(
+                "start",
+                "2026-07-22T10:00:00Z",
+                "session_start",
+                json!({"cwd":"/work/synty"}),
+            ),
+            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 exited with code 0"}),
+            ),
+        ];
+        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()]),
+            cached: None,
+        };
+        let out = backend
+            .list(
+                "spans",
+                None,
+                None,
+                None,
+                None,
+                None,
+                false,
+                Some("2026-07-22T10:00:00Z"),
+                None,
+                "recent",
+                20,
+                &ReadScope::default(),
+            )
+            .unwrap();
+        assert!(out.contains("exec_command"));
+        assert!(out.contains("call-1"));
+        assert!(
+            calls
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|sql| sql.starts_with("SELECT "))
+        );
+    }
+
+    #[test]
+    fn literal_search_reads_matches_and_metadata_without_expanding_whole_sessions() {
+        let lines = vec![
+            event(
+                "start",
+                "2026-07-22T10:00:00Z",
+                "session_start",
+                json!({"cwd":"/work/synty"}),
+            ),
+            event(
+                "result-1",
+                "2026-07-22T10:00:03Z",
+                "tool_result",
+                json!({"call_id":"c1","output":"missing libxcb.so.1"}),
+            ),
+        ];
+        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()]),
+            cached: None,
+        };
+
+        let out = backend
+            .search(
+                "libxcb.so.1",
+                None,
+                None,
+                None,
+                Some("tool_result"),
+                20,
+                &ReadScope::default(),
+            )
+            .unwrap();
+
+        assert!(out.contains("missing libxcb.so.1"));
+        assert!(
+            backend.cached.is_none(),
+            "partial search stores must not satisfy a later trace_show"
+        );
+        let calls = calls.lock().unwrap();
+        assert_eq!(calls.len(), 2, "match query plus metadata query");
+        assert!(calls[0].contains("strpos(lower(line), 'libxcb.so.1') > 0"));
+        assert!(calls[0].contains("$.kind"));
+        assert!(calls[0].contains("'tool_result'"));
+        assert!(calls[1].contains("'session_start', 'agent_meta'"));
+        assert!(calls[1].contains("row_number() OVER"));
+    }
+
+    #[test]
+    fn show_expands_the_resolved_session_in_one_follow_up_query() {
+        let lines = vec![
+            event(
+                "start",
+                "2026-07-22T10:00:00Z",
+                "session_start",
+                json!({"cwd":"/work/synty"}),
+            ),
+            event(
+                "call-1",
+                "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 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()]),
+            cached: None,
+        };
+
+        let out = backend.show("call-1", 3, 5, &ReadScope::default()).unwrap();
+
+        assert!(out.contains("cargo test"));
+        let calls = calls.lock().unwrap();
+        assert_eq!(calls.len(), 2, "id lookup plus one full-session query");
+        assert!(calls[1].contains("$.session_id"));
+        assert!(!calls[1].contains("synty_context_rank"));
+    }
+
+    #[test]
+    fn windows_fail_closed_before_scanning_more_than_seven_days() {
+        let error = Window::parse(
+            Some("2026-07-01T00:00:00Z"),
+            Some("2026-07-09T00:00:00Z"),
+            DEFAULT_LIST_HOURS,
+        )
+        .err()
+        .unwrap();
+        assert!(error.to_string().contains("168 hours"));
+    }
+}

From ac68b8f11681f9353f3a8eca93e8809afa62d1ce Mon Sep 17 00:00:00 2001
From: Attila Laszlo Nagy <attilalaszlo.nagy@superlinked.com>
Date: Fri, 24 Jul 2026 01:13:06 +0200
Subject: [PATCH 2/3] CI: Match quoted Athena Helm arguments

Helm renders the workgroup as a quoted shell argument. Assert the exact safe rendering so the deploy job verifies the intended value instead of failing on the missing unquoted form.
---
 .github/workflows/ci.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3cbfa9c..919c655 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -85,7 +85,7 @@ jobs:
           grep -q 'path: /ready' /tmp/synty-options.yaml
           grep -q 'scheme: HTTPS' /tmp/synty-options.yaml
           grep -q '203.0.113.10/32' /tmp/synty-options.yaml
-          grep -q -- '--athena-workgroup synty-mcp-readonly' /tmp/synty-options.yaml
+          grep -q -- '--athena-workgroup "synty-mcp-readonly"' /tmp/synty-options.yaml
           grep -q '198.51.100.20/32' /tmp/synty-options.yaml
           if grep -q -- '- {}' /tmp/synty-options.yaml; then
             echo "MCP rendered an unrestricted NetworkPolicy rule" >&2

From d29248cc770bcf9cfc1c9509bbb0a265325c4842 Mon Sep 17 00:00:00 2001
From: Attila Laszlo Nagy <attilalaszlo.nagy@superlinked.com>
Date: Fri, 24 Jul 2026 01:28:31 +0200
Subject: [PATCH 3/3] M10: Harden Athena startup and row handling

Reject configured Athena mode before HTTP startup when its backend or required bucket is unavailable. Keep remote readiness independent of the retired trace snapshot, and tolerate malformed raw rows while discovering otherwise valid sessions. Adds three scenarios; the full feature suite passes 298 tests.
---
 src/mcp_http.rs     | 47 +++++++++++++++++++++++++++++++++++++++++++++
 src/readmodel.rs    | 30 ++++++++++++++++++++++++++++-
 src/trace_athena.rs | 21 ++++++++++++++------
 3 files changed, 91 insertions(+), 7 deletions(-)

diff --git a/src/mcp_http.rs b/src/mcp_http.rs
index f756d2f..286d92d 100644
--- a/src/mcp_http.rs
+++ b/src/mcp_http.rs
@@ -211,6 +211,7 @@ pub struct Opts {
 
 #[cfg(feature = "mcp-http")]
 pub fn run(opts: Opts) -> Result<()> {
+    validate_athena_startup(opts.bucket.as_deref(), opts.athena.as_ref())?;
     anyhow::ensure!(opts.token.len() >= 32, "HTTP MCP bearer token must contain at least 32 bytes");
     let tls = tls_config(opts.tls_cert.as_deref(), opts.tls_key.as_deref())?;
     validate_listener(&opts.bind, opts.listen_public, tls.is_some())?;
@@ -250,6 +251,31 @@ pub fn run(opts: Opts) -> Result<()> {
     runtime.block_on(serve(opts.bind, scheme, tls, state))
 }
 
+#[cfg(any(feature = "mcp-http", test))]
+fn validate_athena_startup(
+    bucket: Option<&str>,
+    athena: Option<&crate::mcp::AthenaTraceOptions>,
+) -> Result<()> {
+    let Some(athena) = athena else { return Ok(()) };
+    #[cfg(feature = "athena")]
+    {
+        let bucket = bucket
+            .ok_or_else(|| anyhow::anyhow!("--athena-workgroup requires --bucket s3://..."))?;
+        crate::trace_athena::Config::new(
+            bucket.to_string(),
+            athena.workgroup.clone(),
+            athena.database.clone(),
+            athena.table.clone(),
+        )
+        .map(|_| ())
+    }
+    #[cfg(not(feature = "athena"))]
+    {
+        let _ = (bucket, athena);
+        bail!("Athena trace is not in this binary; rebuild with --features athena")
+    }
+}
+
 #[cfg(feature = "mcp-http")]
 async fn serve(
     bind: String,
@@ -666,6 +692,27 @@ mod tests {
         assert!(validate_listener("0.0.0.0:8765", true, true).is_ok());
     }
 
+    #[test]
+    fn athena_configuration_fails_before_server_start_without_a_backend() {
+        let options = crate::mcp::AthenaTraceOptions {
+            workgroup: "synty-mcp-readonly".into(),
+            database: "synty".into(),
+            table: "raw_events".into(),
+        };
+        #[cfg(not(feature = "athena"))]
+        assert!(
+            validate_athena_startup(Some("s3://team"), Some(&options))
+                .unwrap_err()
+                .to_string()
+                .contains("--features athena")
+        );
+        #[cfg(feature = "athena")]
+        {
+            assert!(validate_athena_startup(Some("s3://team"), Some(&options)).is_ok());
+            assert!(validate_athena_startup(None, Some(&options)).is_err());
+        }
+    }
+
     #[test]
     fn browser_origins_are_exact_allowlist_matches() {
         let allowed = vec!["https://memory.example.com".to_string()];
diff --git a/src/readmodel.rs b/src/readmodel.rs
index 44c3011..f74bb4b 100644
--- a/src/readmodel.rs
+++ b/src/readmodel.rs
@@ -125,7 +125,12 @@ pub fn trace_path() -> PathBuf {
 /// require the published trace snapshot.
 #[cfg(feature = "mcp-http")]
 pub fn mediated_ready(require_trace: bool) -> bool {
-    current().is_some_and(|current| {
+    mediated_ready_from(current(), require_trace)
+}
+
+#[cfg(feature = "mcp-http")]
+fn mediated_ready_from(current: Option<Current>, require_trace: bool) -> bool {
+    current.is_some_and(|current| {
         current.format >= FORMAT
             && current.analysis().is_file()
             && (!require_trace || current.trace().is_file())
@@ -248,6 +253,29 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "mcp-http")]
+    #[test]
+    fn mediated_readiness_requires_trace_only_for_projection_mode() {
+        let dir =
+            std::env::temp_dir().join(format!("synty-mediated-ready-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        std::fs::write(dir.join("analysis.json"), b"{}").unwrap();
+        let current = Current {
+            build: dir.to_string_lossy().into_owned(),
+            rev: 0,
+            format: FORMAT,
+            writer: String::new(),
+        };
+
+        assert!(mediated_ready_from(Some(current.clone()), false));
+        assert!(!mediated_ready_from(Some(current.clone()), true));
+        std::fs::write(dir.join("trace.json"), b"{}").unwrap();
+        assert!(mediated_ready_from(Some(current), true));
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     fn clone_build_copies_a_tree() {
         let dir = std::env::temp_dir().join(format!("synty-clone-{}", std::process::id()));
diff --git a/src/trace_athena.rs b/src/trace_athena.rs
index 3482705..ad11edd 100644
--- a/src/trace_athena.rs
+++ b/src/trace_athena.rs
@@ -383,7 +383,7 @@ impl Backend {
             ..Default::default()
         };
         let first = self.select(&streams, window, &predicate)?;
-        let sessions = event_sessions(&first.lines)?;
+        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"
@@ -640,16 +640,16 @@ fn select_sql(
     }
 }
 
-fn event_sessions(lines: &[String]) -> Result<BTreeSet<String>> {
+fn event_sessions(lines: &[String]) -> BTreeSet<String> {
     let mut sessions = BTreeSet::new();
     for line in lines {
-        let event: crate::event::Event =
-            serde_json::from_str(line).context("parse Athena event row")?;
-        if !event.session_id.is_empty() {
+        if let Ok(event) = serde_json::from_str::<crate::event::Event>(line)
+            && !event.session_id.is_empty()
+        {
             sessions.insert(event.session_id);
         }
     }
-    Ok(sessions)
+    sessions
 }
 
 fn sql_string(value: &str) -> String {
@@ -908,4 +908,13 @@ mod tests {
         .unwrap();
         assert!(error.to_string().contains("168 hours"));
     }
+
+    #[test]
+    fn malformed_rows_do_not_abort_session_discovery() {
+        let sessions = event_sessions(&[
+            "{not-json".into(),
+            event("valid", "2026-07-22T10:00:00Z", "user_prompt", json!({})),
+        ]);
+        assert_eq!(sessions, BTreeSet::from(["session-1".into()]));
+    }
 }