diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index cdba193..919c655 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..286d92d 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,16 +206,19 @@ 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")]
 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())?;
     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 +226,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 +242,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()
@@ -227,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,
@@ -364,14 +413,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 +561,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")
@@ -624,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()];
@@ -645,6 +734,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 +813,7 @@ mod tests {
                 crate::redact::Profile::Off,
                 false,
                 None,
+                None,
             ),
         )
         .unwrap();
diff --git a/src/readmodel.rs b/src/readmodel.rs
index f326ac5..f74bb4b 100644
--- a/src/readmodel.rs
+++ b/src/readmodel.rs
@@ -120,13 +120,20 @@ 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 {
-    current().is_some_and(|current| {
-        current.format >= FORMAT && current.analysis().is_file() && current.trace().is_file()
+pub fn mediated_ready(require_trace: bool) -> bool {
+    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())
     })
 }
 
@@ -246,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/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..ad11edd
--- /dev/null
+++ b/src/trace_athena.rs
@@ -0,0 +1,920 @@
+// 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]) -> BTreeSet<String> {
+    let mut sessions = BTreeSet::new();
+    for line in lines {
+        if let Ok(event) = serde_json::from_str::<crate::event::Event>(line)
+            && !event.session_id.is_empty()
+        {
+            sessions.insert(event.session_id);
+        }
+    }
+    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"));
+    }
+
+    #[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()]));
+    }
+}