diff --git a/CHANGELOG.md b/CHANGELOG.md
index efa1f09..ae11176 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,15 +17,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Filter.** With either limit set, `capture sync` promotes only captures
   recorded on `defaults.source_model`, and with `max_age` set, only captures
   recorded within it. Model ids match on bare names: lower-cased, provider
-  prefix and trailing snapshot date dropped, no alias lookup. `max_age` alone
-  caps nothing. `capture promote` ignores the limits.
+  prefix and trailing snapshot date dropped, no alias lookup. A conversation
+  turn that recorded no messages list of its own is promoted only when every
+  earlier turn of its conversation is local, since its history is rebuilt
+  from them (`not eligible: N missing earlier turns`, with a warning naming
+  the likely causes: the fetch window or budget, `capture clean`, the SDK's
+  per-turn sample rate, or a `turn_index` numbered from 1). `max_age` alone caps
+  nothing. `capture promote` ignores the limits.
 - **Plan.** With `max_cases` set, sync never grows a suite past the cap and
   swaps stale cases (too old, or recorded on another model) for fresh
   captures as they arrive. A stale case leaves at once when a fresh capture of
   the same question is added in its place; if its fresh copy is not added,
   the stale case leaves only to make room, like any other stale case. So a
   suite never shrinks because its cases went stale (an over-cap suite is still
-  trimmed to `max_cases`). A conversation enters whole or not at all. Cases a
+  trimmed to `max_cases`). A conversation enters whole when it fits; once no
+  whole conversation fits, the slots left take the next one's leading turns,
+  so one long conversation cannot leave a suite nearly empty. Cases a
   human marked `provenance: reviewed` are never retired; in a captured suite
   the mark goes in the case file (`.evalshift/suites/<suite>/<case>.json`),
   not in `golden.jsonl`, which sync regenerates. Sync knows a case by its
@@ -43,13 +50,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   recorded on <model> …; not filtering by model — check
   defaults.source_model`) and does not filter that suite by model rather than
   empty it.
-- **Fetch.** `max_age` also bounds which captures `capture sync`'s automatic
-  fetch downloads from `captures.store` (toolset sidecars are always
-  fetched); `--since` still overrides it, and `capture list` and
-  `capture fetch` are unchanged.
+- **Fetch.** The limits also bound what the automatic fetch of
+  `capture sync` and `capture list` downloads from `captures.store`:
+  `max_age` skips older captures, and `max_cases` takes only each suite's
+  newest `3 × max_cases` captures inside that window (`skipped 2400 beyond
+  each suite's newest 600`). Captures already local or promoted count toward
+  that budget, so each sync downloads at most that many per suite, however
+  many captures the bucket holds, and a re-sync with nothing new downloads
+  nothing. Toolset sidecars are always fetched; `--since` replaces the age
+  cutoff, not the budget; `capture fetch` is unchanged.
 - **Output.** Sync prints one line per suite: `alpha: 200 cases — kept 150,
   retired 50 (32 older than 14d, 18 on another model), added 50 of 2940
-  eligible; not eligible: …`.
+  eligible; not eligible: …`. With `max_cases` set, a capture's build
+  warnings print only when it becomes a case.
 - `doctor` gains `suite freshness` rows for each `source: captured` suite when
   either limit is set: `ok` (`alpha: 200 cases, captured 2d–12d ago`), or one
   `warn` per finding — the newest case is older than `max_age`, cases were
diff --git a/DOCS.md b/DOCS.md
index 6b906d2..c502678 100644
--- a/DOCS.md
+++ b/DOCS.md
@@ -177,7 +177,7 @@ captures:
 
 and `capture sync` and `capture list` fetch what is new before they read the local directory — captures are immutable and toolsets content-addressed, so only objects missing locally are downloaded, captures already promoted are skipped (so `capture clean` never causes a re-download; `capture sync --repair` alone re-fetches the promoted captures its rebuilds need, past `--since` too), and each file is written atomically. Bucket keys are untrusted: one not shaped like the SDK's layout — a suite directory containing `:`, `/`, `\`, NUL or `..`, or starting or ending with `.` or a space, included — is counted as malformed and never written (a key with a trailing newline included), and no key can place a file outside `.evalshift/`; folder-marker keys ending in `/` (the S3 console's "Create folder", ADLS directories) are skipped without being counted. One summary line reports it (`fetched 12 capture(s) and 1 toolset(s) from s3://…`; with `capture list --json` it goes to stderr, so stdout stays pure JSON). `--since 24h` limits a fetch to recently written captures, `--offline` skips it, and `capture fetch` does only that step. `promote`, `diff` and `clean` stay local. Install the matching client library — `pip install boto3`, `google-cloud-storage` or `azure-storage-blob azure-identity` — and authenticate with the provider's normal chain (an OIDC role in CI, `aws sso login` / `gcloud auth application-default login` / `az login` locally); credentials never go in the URI, and a value containing `@` or `?` fails to load; no rejected value is echoed back, at most its scheme. A missing client library exits 1 naming the `pip install <package>` to run; a list or download failure exits 1, so `sync` promotes nothing from a partial fetch, with a hint naming the provider's credential chain (`aws sso login` / `AWS_PROFILE` / the CI OIDC role, `gcloud auth application-default login`, `az login`); a local write failure names the file and asks you to check free disk space and permissions instead. If `evalshift.yaml` cannot be read or does not validate, `sync` and `list` skip the fetch and work on the local directory as before, warning only when the file has a top-level `captures:` key. Full field reference: [`captures`](docs/configuration.md#captures).
 
-**Bounded suites.** Together, `captures.max_age` (a duration: `30m`, `24h`, `14d`, at most `36500d`; omit it for no age limit) and `captures.max_cases` (an int ≥ 1), both optional, keep each captured suite a capped, route-balanced sample of recent production on `defaults.source_model`; `evalshift init` scaffolds `max_age: 14d` and `max_cases: 200`. Setting either one turns the feature on: `max_age` filters what enters a suite and marks old cases stale, and `max_cases` caps and route-balances each suite. Without them `capture sync` promotes and keeps exactly what it did before, and `defaults.source_model` alone changes nothing. Case files record `captured_at` and `source_model` (the first `model_call`'s `model_id`) either way, and sync backfills both on existing cases whose capture is still local (`backfilled capture time and model on N case(s)`). With the limits set, sync promotes only captures recorded within `max_age` on `defaults.source_model`. Models match on bare names: lower-cased, provider prefix and one trailing snapshot date (`-20260301`, `-2026-08-07`) dropped, with no alias lookup, so set `defaults.source_model` to the id your app sends. If none of a suite's cases and captures was recorded on it, sync warns (`alpha: none of 212 case(s) and capture(s) were recorded on claude-sonnet-5-5 (seen: …); not filtering by model — check defaults.source_model`) and does not filter that suite by model, so a typo or a rollout whose first new-model capture has not arrived can neither empty nor freeze a suite. `max_cases` caps every suite under `.evalshift/suites/`, including ones promoted by hand with `capture promote` and ones whose entry is `managed: false` (that flag freezes the YAML entry, not the case files): new conversations enter whole or not at all, least-represented tool route first, in a fixed hash order, so the same captures give the same suite on any machine. A stale case (too old, or recorded on another model) is retired outright when a fresh capture of the same question is added in its place; if its fresh copy is not added, the stale case leaves only to make room, like any other stale case. Stale cases that leave to make room go other-model first, then oldest, so a suite never shrinks because it went stale. A suite still over the cap loses fresh cases from its most-represented route, oldest first. Cases marked `provenance: reviewed` are never retired; in a captured suite, mark them in the case file `.evalshift/suites/<suite>/<case>.json` (on the calls under `example.expected_tools`, and `example.expected_tool_rounds` when present), not in `golden.jsonl`, which sync regenerates from the case files on every run. `max_age` also bounds `capture sync`'s automatic fetch (`skipped 11 older than captures.max_age (14d)`); `--since` overrides it, and `capture list` and `capture fetch` do not apply it. On a fresh checkout or after `capture clean`, a conversation that straddles the window loses its earlier turns to it, and a turn that does not record its own messages list then rebuilds a truncated history: pass the full messages list to `capture.model_call`. `doctor` and `compare` report stale suites (below). Expire the bucket's `captures/` prefix with a lifecycle rule after a period ≥ `max_age`; the CLI never deletes from the bucket. Full rules, output and lifecycle examples: [Bounded suites](docs/configuration.md#bounded-suites).
+**Bounded suites.** Together, `captures.max_age` (a duration: `30m`, `24h`, `14d`, at most `36500d`; omit it for no age limit) and `captures.max_cases` (an int ≥ 1), both optional, keep each captured suite a capped, route-balanced sample of recent production on `defaults.source_model`; `evalshift init` scaffolds `max_age: 14d` and `max_cases: 200`. Setting either one turns the feature on: `max_age` filters what enters a suite and marks old cases stale, and `max_cases` caps and route-balances each suite. Without them `capture sync` promotes and keeps exactly what it did before, and `defaults.source_model` alone changes nothing. Case files record `captured_at` and `source_model` (the first `model_call`'s `model_id`) either way, and sync backfills both on existing cases whose capture is still local (`backfilled capture time and model on N case(s)`). With the limits set, sync promotes only captures recorded within `max_age` on `defaults.source_model`. Models match on bare names: lower-cased, provider prefix and one trailing snapshot date (`-20260301`, `-2026-08-07`) dropped, with no alias lookup, so set `defaults.source_model` to the id your app sends. If none of a suite's cases and captures was recorded on it, sync warns (`alpha: none of 212 case(s) and capture(s) were recorded on claude-sonnet-5-5 (seen: …); not filtering by model — check defaults.source_model`) and does not filter that suite by model, so a typo or a rollout whose first new-model capture has not arrived can neither empty nor freeze a suite. `max_cases` caps every suite under `.evalshift/suites/`, including ones promoted by hand with `capture promote` and ones whose entry is `managed: false` (that flag freezes the YAML entry, not the case files): new conversations enter whole, least-represented tool route first, in a fixed hash order, so the same captures give the same suite on any machine; once no whole conversation fits, the slots left take the next one's leading turns. A conversation turn that recorded no messages list of its own rebuilds its history from the turns before it, so under the limits it is promoted only when every earlier turn is local (`not eligible: 3 missing earlier turns`). A stale case (too old, or recorded on another model) is retired outright when a fresh capture of the same question is added in its place; if its fresh copy is not added, the stale case leaves only to make room, like any other stale case. Stale cases that leave to make room go other-model first, then oldest, so a suite never shrinks because it went stale. A suite still over the cap loses fresh cases from its most-represented route, oldest first. Cases marked `provenance: reviewed` are never retired; in a captured suite, mark them in the case file `.evalshift/suites/<suite>/<case>.json` (on the calls under `example.expected_tools`, and `example.expected_tool_rounds` when present), not in `golden.jsonl`, which sync regenerates from the case files on every run. The limits also bound the automatic fetch of `capture sync` and `capture list`: `max_age` skips older captures (`skipped 11 older than captures.max_age (14d)`), and `max_cases` downloads only each suite's newest `3 × max_cases` captures inside that window (`2400 beyond each suite's newest 600`), counting captures already local or promoted, so each sync downloads at most that many per suite and a re-sync with nothing new downloads nothing. `--since` replaces the age cutoff, not the budget, and `capture fetch` applies neither. A turn without its own messages list is also missing earlier turns after `capture clean`, on a fresh checkout (promoted turns are never fetched), when `EVALSHIFT_SAMPLE_RATE` (drawn per turn) dropped them, or when `turn_index` starts at 1 instead of 0; sync warns naming these causes. Pass the full messages list to `capture.model_call` so each turn carries its own history. With `max_cases` set, build warnings print only for captures that become cases. `doctor` and `compare` report stale suites (below). Expire the bucket's `captures/` prefix with a lifecycle rule after a period ≥ `max_age`; the CLI never deletes from the bucket. Full rules, output and lifecycle examples: [Bounded suites](docs/configuration.md#bounded-suites).
 
 **`capture sync`** is the workhorse. Per suite it:
 
diff --git a/docs/configuration.md b/docs/configuration.md
index 0acce8e..1917a83 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1086,8 +1086,8 @@ captures:
 | Field       | Type                              | Default | Description |
 | ----------- | --------------------------------- | ------- | ----------- |
 | `store`     | string (URI)                      | (none)  | Object store the SDK writes to. `capture sync` and `capture list` fetch new captures and toolset sidecars from it into the local directory first; `capture fetch` does only that step. |
-| `max_age`   | duration (`30m`, `24h`, `14d`; > 0, ≤ `36500d`) | (none)  | Captures recorded longer ago than this never enter a suite through `capture sync` (`capture promote` ignores the limits), and promoted cases that old are stale: with `max_cases` also set, `capture sync` replaces them as fresh captures arrive. Also the window `capture sync` fetches from `store`. Omit for no age limit. |
-| `max_cases` | int (≥ 1)                         | (none)  | The most cases `capture sync` keeps in each captured suite. Omit and suites grow with every sync. |
+| `max_age`   | duration (`30m`, `24h`, `14d`; > 0, ≤ `36500d`) | (none)  | Captures recorded longer ago than this never enter a suite through `capture sync` (`capture promote` ignores the limits), and promoted cases that old are stale: with `max_cases` also set, `capture sync` replaces them as fresh captures arrive. Also the window `capture sync` and `capture list` fetch from `store`. Omit for no age limit. |
+| `max_cases` | int (≥ 1)                         | (none)  | The most cases `capture sync` keeps in each captured suite. Also caps what `capture sync` and `capture list` download from `store`: each suite's newest `3 × max_cases` captures per run. Omit and suites grow with every sync. |
 
 Setting either `max_age` or `max_cases` turns [bounded suites](#bounded-suites) on. Both work
 without `store`: they govern how captures become suites, whether the captures came from disk or
@@ -1122,7 +1122,9 @@ already promoted into a suite (so `capture clean` never causes a re-download;
 `capture sync --repair` alone re-fetches the promoted captures its rebuilds need, past `--since`
 too), and writes each file atomically. `--since 24h` (or `7d`, `30m`, an ISO date) limits a fetch to recently
 written captures; `--offline` skips it and works with what is already local. With `max_age` set,
-`capture sync`'s own fetch skips captures older than it (see [Fetch window](#fetch-window)).
+the fetch that `capture sync` and `capture list` run skips captures older than it, and with
+`max_cases` set it takes only each suite's newest `3 × max_cases` (see
+[Fetch window](#fetch-window)).
 
 Retention in the bucket is yours to set with a lifecycle rule; the SDK's `max_captures` does not
 apply there, and the CLI never writes to or deletes from the bucket. Expire the `captures/` prefix
@@ -1173,7 +1175,12 @@ case. The summary line then ends `, backfilled capture time and model on N case(
   `defaults.source_model` to the id your app sends, not a friendly alias for it, or the guardrail
   below trips.
 - **Eligible**: a capture not yet promoted that survives the usual filters (no events, blocked,
-  errored, duplicate) and is neither too old nor on another model.
+  errored, duplicate), is neither too old nor on another model, and is not **missing earlier
+  turns**. A conversation turn that did not record its own messages list rebuilds its history
+  from the turns before it, so every lower `turn_index` (0-based) of its conversation must be on
+  disk; one cut off by the [fetch window](#fetch-window) or budget, or promoted on another
+  machine and so never fetched, makes the turn ineligible rather than promoted with a wrong
+  history.
 
 **Guardrail.** For each suite, when at least one of its cases and new captures recorded a model
 and none of them was recorded on `defaults.source_model`, sync does not filter that suite by
@@ -1188,13 +1195,17 @@ empty a suite nor freeze it.
 
 **The plan** (with `max_cases` set). Sync plans each suite in four steps:
 
-1. **A conversation is the unit.** The new turns of one conversation enter together or not at
-   all; a conversation larger than the free slots is skipped and the next one tried. A capture
-   without a `conversation_id` is its own unit.
+1. **A conversation is the unit.** The new turns of one conversation enter together when they
+   fit; a conversation larger than the free slots is skipped and the next one tried. Once no whole
+   conversation fits, the slots left take the leading turns of the next conversation in the order
+   below, so one long conversation cannot leave a suite nearly empty. Its later turns can enter
+   on a later sync while the earlier turns' captures are still local; after `capture clean` or
+   on a fresh checkout, a later turn that does not record its own messages list is missing
+   earlier turns (below). A capture without a `conversation_id` is its own unit.
 2. **Fill free slots, route-balanced.** The free slots are `max_cases` minus the reviewed and fresh
    cases already in the suite. A route is the set of tool names a case expects in round 1 (no
    tools is a route too). Each pick goes to the least-represented route that still has a
-   conversation that fits. Within a route, conversations are taken in the order of a hash of their
+   conversation that fits (for the leading turns above, the least-represented route left). Within a route, conversations are taken in the order of a hash of their
    id, with no randomness: the same captures give the same suite on any machine, and syncing again
    with nothing new changes no file.
 3. **Retire stale cases to make room, not because they are stale.** A stale case whose question an
@@ -1259,27 +1270,46 @@ and a fresh capture of a stale case's question is skipped as a duplicate.
 
 `kept` counts the cases that stay, `retired` the ones that left (with `N over max_cases` for a
 trim), `added N of M eligible` the new cases out of the eligible captures, and `not eligible` the
-captures filtered as too old or on another model. Zero counts are left out. With `max_age` alone
-the line is only `  planner: not eligible: 812 older than 14d`, printed when something was
-filtered.
+captures filtered as too old, on another model, or missing earlier turns. Zero counts are left
+out. With `max_age` alone the line is only `  planner: not eligible: 812 older than 14d`, printed
+when something was filtered. With `max_cases` set, a capture's build warnings (a lost system
+prompt, later rounds) print only when it becomes a case, not for every eligible capture.
 
 #### Fetch window
 
-With `max_age` set and `store` configured, the automatic fetch that `capture sync` runs skips
-captures last written more than `max_age` ago. Toolset sidecars missing locally are always
-fetched, whatever their age. The summary names the cutoff:
+With `store` configured, the automatic fetch that `capture sync` and `capture list` run applies
+the limits too, using the time the bucket last wrote each object:
+
+- **`max_age`** skips captures last written more than `max_age` ago.
+- **`max_cases`** downloads only each suite's newest `3 × max_cases` captures inside that window
+  (ties by key): at most `3 × max_cases` downloads per suite per sync, however many captures the
+  bucket holds, and enough spare for the plan to balance tool routes. Captures already local or
+  promoted count toward it, so syncing again with nothing new downloads nothing. It bounds each
+  sync, not the local mirror: frequent syncs each take the newest captures since the last one,
+  and `capture clean` prunes the mirror.
+
+The summary names both:
 `fetched 40 capture(s) and 0 toolset(s) from s3://… (skipped 4 promoted, 11 older than
-captures.max_age (14d))`. `--since` overrides it. `capture list` and `capture fetch` do not apply
-it. The fetch does not sample: a listing gives only key, time and size, and a turn without its own
-messages list rebuilds its history from sibling captures, so everything inside the window is
-downloaded. `EVALSHIFT_SAMPLE_RATE` on the agent governs the volume. On a fresh checkout, the
-committed case files tell the fetch which captures are already promoted, so only new ones inside
-the window are downloaded.
-
-On a fresh checkout or after `capture clean`, a conversation that straddles the `max_age` window
-loses its earlier turns to the fetch window. A later turn that does not record its own messages
-list then rebuilds a truncated history from the turns it has. Pass the full messages list to
-`capture.model_call` so each turn carries its own history.
+captures.max_age (14d), 2400 beyond each suite's newest 600)`. Toolset sidecars missing locally
+are always fetched, whatever their age. `--since` replaces the `max_age` cutoff, not the budget.
+`capture fetch` applies neither: it mirrors everything, or what `--since` asks for.
+`capture sync --repair` fetches the captures its rebuilds need beyond both. On a fresh checkout,
+the committed case files tell the fetch which captures are already promoted.
+
+The budget takes the newest captures, not a sample of the window: at high volume a fresh
+checkout's suite comes from the most recent stretch of traffic. `EVALSHIFT_SAMPLE_RATE` on the
+agent bounds the volume at the source, but it draws once per agent run, that is per turn, so a
+sampled conversation keeps some turns and loses others: with a sample rate, record the full
+messages list on every turn.
+
+A conversation turn that does not record its own messages list rebuilds its history from the
+turns before it, so it is not eligible unless all of them are on disk
+(`not eligible: 3 missing earlier turns`). They are not when the window or the budget cut them
+off, when they were promoted on another machine (a fresh checkout never fetches promoted
+captures) or removed by `capture clean`, when the SDK's sample rate dropped them, or when the
+app numbers `turn_index` from 1 instead of 0. Sync then warns with those causes
+(`alpha: 3 turn(s) not promoted: earlier turns of their conversation are not on disk (…)`).
+Pass the full messages list to `capture.model_call` so each turn carries its own history.
 
 If `evalshift.yaml` cannot be read or does not validate, sync runs without the limits. When the
 file has a top-level `captures:` key, it says so, online or `--offline`:
diff --git a/docs/conversations.md b/docs/conversations.md
index f8f6058..c9283ba 100644
--- a/docs/conversations.md
+++ b/docs/conversations.md
@@ -92,8 +92,16 @@ one turn in the conversation recorded a full messages list. Prefer the
 messages-list convention above when you can.
 
 With `captures.max_cases` set, a conversation's new turns enter a suite
-together or not at all: one larger than the suite's free slots is skipped,
-never split (see [Bounded suites](configuration.md#bounded-suites)).
+together when they fit; one larger than the suite's free slots is skipped
+for the next one. Once no whole conversation fits, the slots left take the
+leading turns of the next conversation; its remaining turns can enter on a
+later sync. Under the limits, a turn that recorded no messages list of its
+own is promoted only when every earlier turn of its conversation is local
+(its history is rebuilt from them), so after `capture clean` or on a fresh
+checkout, where promoted turns are not local, only turns carrying their own
+messages list can still enter. `turn_index` is 0-based: numbering from 1
+makes every rebuilt turn miss its "turn 0" (see
+[Bounded suites](configuration.md#bounded-suites)).
 
 ```bash
 evalshift capture sync                              # promote captures,
diff --git a/docs/sdk.md b/docs/sdk.md
index 749770c..ae08649 100644
--- a/docs/sdk.md
+++ b/docs/sdk.md
@@ -141,7 +141,7 @@ route-balanced sample of recent captures on `defaults.source_model`, and replace
 fresh captures arrive (see [Bounded suites](configuration.md#bounded-suites)).
 
 ```bash
-evalshift capture list                  # what was recorded (--json for machine output)
+evalshift capture list                  # what was recorded, fetched within the limits (--json)
 evalshift capture fetch                 # mirror new captures from captures.store
 evalshift capture sync                  # promote captures → suites + wire evalshift.yaml
 evalshift capture promote cap_ab12 --as refund_case_1
diff --git a/docs/superpowers/specs/2026-10-16-bounded-capture-suites-design.md b/docs/superpowers/specs/2026-10-16-bounded-capture-suites-design.md
index 95e0abd..5a1c10d 100644
--- a/docs/superpowers/specs/2026-10-16-bounded-capture-suites-design.md
+++ b/docs/superpowers/specs/2026-10-16-bounded-capture-suites-design.md
@@ -8,6 +8,12 @@
 > (§3, §6, §7, §8) were written before implementation and are not the CLI's exact text. The
 > source and the user docs (`docs/configuration.md#bounded-suites`, `DOCS.md`, `llms-full.txt`)
 > are authoritative.
+>
+> **Amended 2026-10-19, after the first real-bucket sync** (52 captures, `max_cases: 10`): a lone
+> capture took 1 slot and a 10-turn conversation no longer fit the 9 left, so the suite held 1
+> case. §4 gains the download cap it listed as a follow-up, `capture list` now fetches what sync
+> does, and §5 fills slots no whole group fits with a group's leading turns. Under the limits, a
+> turn whose rebuilt history would miss earlier turns is not eligible (§3).
 
 ## Problem
 
@@ -121,7 +127,9 @@ A wrong or unrecognised `source_model` can therefore never empty or freeze a sui
 holds mid-rollout: until at least one new-model capture exists, nothing is filtered.
 
 **Eligible candidate** = a local capture that is not promoted, survives today's filters (events,
-blocked, errored, dedup — §6), and is neither too old nor on another model.
+blocked, errored, dedup — §6), and is neither too old nor on another model. *(Amended:)* nor
+missing earlier turns — a conversation turn without its own messages list needs every lower
+`turn_index` of its conversation on disk, or its rebuilt history would be wrong.
 
 ### 4. Fetch is bounded by time, not count
 
@@ -136,6 +144,14 @@ sibling captures on disk (`promote.py:1369-1399`), so a key-level sample would s
 histories. Volume inside the window is governed by `EVALSHIFT_SAMPLE_RATE` in the agent. A hard
 download cap is a follow-up.
 
+*(Amended: the cap.)* With `max_cases` set, the fetch takes each suite's newest
+`3 × max_cases` captures inside the cutoff, by `last_modified` (unknown = oldest, ties by key).
+Captures already local or promoted count toward it, so a re-run with nothing new downloads
+nothing. Forced (`--repair`) ids bypass it; `--since` replaces the cutoff, not the cap. Newest-first
+keeps a conversation's turns together except at the cap's edge, where the turns cut off make the
+later ones "missing earlier turns" (§3). `capture list`'s fetch applies the same cutoff and cap;
+`capture fetch` neither.
+
 ### 5. The per-suite plan (only when `max_cases` is set)
 
 New pure module `captures/selection.py` — no I/O, no clock:
@@ -155,7 +171,10 @@ route, content keys. `SuitePlan`: groups to add, `(case name, reason)` to retire
 
 - **Unit = conversation group.** All candidate turns sharing a `conversation_id` within the suite;
   a capture without one is its own group. A group is added whole or not at all; one larger than
-  the remaining slots is skipped and the next tried.
+  the remaining slots is skipped and the next tried. *(Amended:)* once no whole group fits and
+  slots remain, the least-represented route left gives its next group's leading turns to fill
+  them; the rest of that conversation can enter later while its earlier turns' captures are
+  local.
 - **Route** of a case or group = the sorted set of tool names in its round-1 `expected_tools`
   across its turns, `∅` when none.
 - **Order is deterministic, no RNG:** within a route, groups are taken in `sha256(group key)` order.
diff --git a/llms-full.txt b/llms-full.txt
index f158074..d232f68 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -355,7 +355,7 @@ Captures (read from .evalshift/captures/ under CWD, or $EVALSHIFT_DIR; with capt
 in evalshift.yaml, list/sync first mirror the bucket into it):
 evalshift capture list [SUITE] [--json] [-c CONFIG] [--offline]
   With captures.store set, runs fetch first (limited to SUITE when given) unless --offline.
-  captures.max_age does NOT bound list's fetch (only sync's).
+  The fetch applies the same captures.max_age window and max_cases budget as sync's.
   With --json, stdout is pure JSON: the fetch summary and every warning go to stderr.
 evalshift capture fetch [--suite S] [--since 30m|24h|7d|ISO] [-c CONFIG]
   Mirrors captures.store into <base>: lists toolsets/ then captures/, downloads only objects
@@ -376,8 +376,8 @@ evalshift capture fetch [--suite S] [--since 30m|24h|7d|ISO] [-c CONFIG]
   naming pip install boto3 | google-cloud-storage | azure-storage-blob azure-identity (az://
   needs both; one command installs them). No captures.store, or no evalshift.yaml -> exit 1;
   an invalid evalshift.yaml -> its load error, exit 1; an unreadable or non-UTF-8 one -> a one-line
-  "could not read <path>" error, exit 1. Has no --offline flag. captures.max_age does NOT
-  bound it; only sync's automatic fetch applies the max_age window.
+  "could not read <path>" error, exit 1. Has no --offline flag. captures.max_age and the
+  max_cases budget do NOT bound it; only the automatic fetch of sync and list applies them.
 evalshift capture promote CAPTURE_ID [--as CASE_ID] [--suite S] [--input-var NAME=input]
     [--tag T]... [--strict-args] [--names-only] [--tool-count] [--rounds first(default)|all]
     [--allow-errored] [-f/--force]
@@ -477,7 +477,17 @@ evalshift capture sync [--suite S] [--input-var NAME=input] [--tag T]... [--stri
     tried). slots = max_cases - reviewed - fresh; repeatedly add the
     next group of the route (sorted round-1 expected tool names; none is its own route) with the
     fewest cases so far that still has a group that fits (ties by route name); within a route,
-    sha256(group key) order. Retire: a stale case whose content key an added turn replays is
+    sha256(group key) order. Once NO whole group fits and slots remain, the least-represented
+    route left gives its next group's LEADING turns (a prefix, never a middle slice) to fill the
+    slots; the rest of that conversation can enter on a later sync while its earlier turns'
+    captures are local (after capture clean / on a fresh checkout, a rest turn without its own
+    messages list is "missing earlier turns").
+    Eligibility under the limits also needs earlier turns: a conversation turn whose capture
+    recorded no messages list of its own (history rebuilt from sibling turns) is NOT eligible
+    unless every lower turn_index (0-based) of its conversation is local -- reported as
+    "not eligible: N missing earlier turns" plus a per-suite warning naming the causes (cut by
+    max_age/budget, promoted elsewhere or capture clean, SDK sample rate -- it draws PER RUN, i.e.
+    per turn -- or a turn_index numbered from 1). Retire: a stale case whose content key an added turn replays is
     retired OUTRIGHT, even under the cap (no duplicate content; also under --keep-duplicates); if
     its fresh copy is not added, the stale case leaves only to make room, like any other stale
     case. Other stale cases only while over max_cases: model-stale first (oldest first, unknown
@@ -498,16 +508,22 @@ evalshift capture sync [--suite S] [--input-var NAME=input] [--tag T]... [--stri
       added 50 of 2940 eligible; not eligible: 812 older than 14d, 40 on another model"
     (retired reasons also "N over max_cases"; kept = cases that stay; eligible = candidates after
     dedup). max_age alone: "  planner: not eligible: 812 older than 14d", only when non-empty.
-    Fetch window: with max_age set, sync's AUTOMATIC fetch skips CAPTURES whose last_modified is
-    before now - max_age; toolset sidecars missing locally are always fetched ("(skipped 4
-    promoted, 11 older than captures.max_age (14d))"); --since overrides it; capture list /
-    capture fetch do not apply it. No sampling at fetch time (a listing has key/time/size only;
-    history rebuild needs sibling turns) -- volume inside the window is EVALSHIFT_SAMPLE_RATE's
-    job. Committed case files supply the promoted-id skip list, so a fresh checkout downloads only
-    new captures inside the window. On a fresh checkout or after capture clean, a conversation
-    straddling the window loses its earlier turns to it; a turn that does not record its own
-    messages list then rebuilds a TRUNCATED history -- pass the full messages list to
-    capture.model_call.
+    With max_cases set, a capture's build warnings print only if it becomes a case.
+    Fetch window + budget (the AUTOMATIC fetch of capture sync AND capture list): with max_age
+    set, CAPTURES whose last_modified is before now - max_age are skipped; with max_cases set,
+    only each suite's NEWEST 3 x max_cases captures inside that window are downloaded (by
+    last_modified, unknown = oldest, ties by key; captures already local or promoted COUNT toward
+    it, so a re-run with nothing new downloads nothing). Toolset sidecars missing locally are
+    always fetched. Summary: "(skipped 4 promoted, 11 older than captures.max_age (14d), 2400
+    beyond each suite's newest 600)". --since replaces the max_age cutoff, NOT the budget;
+    capture fetch applies neither; sync --repair fetches the captures its rebuilds need beyond
+    both. At most 3 x max_cases downloads per suite PER SYNC (not a bound on the local mirror).
+    The budget is the newest captures, not a sample of the window; EVALSHIFT_SAMPLE_RATE on the
+    agent bounds volume at the source but samples per turn, so sampled conversations need the
+    full messages list on every turn. Committed case files supply the promoted-id skip list.
+    The window/budget can cut a conversation's earlier turns, and on a fresh checkout its
+    promoted turns are never fetched: a turn without its own messages list is then "missing
+    earlier turns" (ineligible) -- pass the full messages list to capture.model_call.
     Route balancing is coverage-weighted, not traffic-weighted: rate budgets are measured over a
     balanced sample. The CLI never writes to or deletes from the bucket; expire the captures/
     prefix (not toolsets/, which committed suites reference) with a lifecycle rule after a period
@@ -1600,7 +1616,8 @@ content leaves the machine.
   not seed it (its fresh re-capture replaces it); every case that stays does.
 - Bounded capture sync (captures.max_age / captures.max_cases set): a suite never shrinks
   because it went stale (stale cases leave only to make room or to be replaced by their fresh
-  twin); reviewed cases are never retired; a conversation is added whole or not at all; the
+  twin); reviewed cases are never retired; a conversation is added whole when it fits, else
+  only as its leading turns once no whole one fits; the
   plan is deterministic, and syncing again with nothing new changes no file; a
   defaults.source_model no capture ran on filters nothing (guardrail warning).
 - Slices: every distinct example tag is a slice under its own name, plus "all"; each evaluator is
diff --git a/src/evalshift_cli/captures/remote.py b/src/evalshift_cli/captures/remote.py
index b65b0d6..a32f31b 100644
--- a/src/evalshift_cli/captures/remote.py
+++ b/src/evalshift_cli/captures/remote.py
@@ -26,7 +26,7 @@
 import os
 import re
 import tempfile
-from collections.abc import Collection, Iterator
+from collections.abc import Collection, Iterator, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import suppress
 from dataclasses import dataclass
@@ -198,7 +198,9 @@ class FetchSummary:
     skipped_old: int = 0
     skipped_large: int = 0
     skipped_malformed: int = 0
+    skipped_budget: int = 0
     since_label: str = "--since"
+    per_suite_limit: int | None = None
 
     def describe(self) -> str:
         """``fetched N capture(s) and M toolset(s) from <uri> (skipped ...)``."""
@@ -212,6 +214,7 @@ def describe(self) -> str:
                 (self.skipped_existing, "already local"),
                 (self.skipped_promoted, "promoted"),
                 (self.skipped_old, f"older than {self.since_label}"),
+                (self.skipped_budget, f"beyond each suite's newest {self.per_suite_limit}"),
                 (self.skipped_large, "over the size cap"),
                 (self.skipped_malformed, "malformed"),
             )
@@ -256,6 +259,7 @@ def fetch_captures(
     force_ids: Collection[str] = frozenset(),
     workers: int = 8,
     since_label: str = "--since",
+    per_suite_limit: int | None = None,
 ) -> FetchSummary:
     """Mirror missing toolsets and captures from ``store`` into ``<base>``.
 
@@ -269,11 +273,18 @@ def fetch_captures(
     match the SDK's layout are counted as malformed and ignored. ``since_label`` is how the
     summary names the ``since`` cutoff.
 
+    ``per_suite_limit`` caps the download, not just the age: only each suite's newest that-many
+    captures inside ``since`` are fetched -- newest by ``last_modified`` (unknown counts as
+    oldest), ties by key. Captures already local or promoted count toward it, so a re-run with
+    nothing new in the store fetches nothing. Forced ids are fetched beyond it.
+
     Raises:
         RemoteStoreError: when listing or any download fails. Files already written stay, so a
             re-run resumes; nothing half-written is left behind.
     """
-    summary = FetchSummary(store_uri=store.uri, since_label=since_label)
+    summary = FetchSummary(
+        store_uri=store.uri, since_label=since_label, per_suite_limit=per_suite_limit
+    )
     skip = set(skip_ids)
     force = set(force_ids)
     toolsets_dir = _absolute(toolsets_root(base))
@@ -300,6 +311,7 @@ def fetch_captures(
 
     capture_jobs: list[tuple[str, Path]] = []
     prefix = f"captures/{suite}/" if suite else "captures/"
+    listed: list[tuple[ObjectInfo, str, Path]] = []
     for info in _list(store, prefix):
         if info.key.endswith("/"):
             continue  # a folder marker, as above
@@ -317,10 +329,18 @@ def fetch_captures(
         if not _absolute(dest.parent).is_relative_to(captures_dir):
             summary.skipped_malformed += 1
             continue
+        listed.append((info, suite_seg, dest))
+
+    in_budget = (
+        _newest_per_suite(listed, since=since, limit=per_suite_limit)
+        if per_suite_limit is not None
+        else None
+    )
+    for info, _, dest in listed:
         if dest.exists():
             summary.skipped_existing += 1
             continue
-        capture_id = name[: -len(".json")]
+        capture_id = dest.name[: -len(".json")]
         forced = capture_id in force
         if not forced and capture_id in skip:
             summary.skipped_promoted += 1
@@ -333,6 +353,9 @@ def fetch_captures(
         ):
             summary.skipped_old += 1
             continue
+        if in_budget is not None and not forced and info.key not in in_budget:
+            summary.skipped_budget += 1
+            continue
         if info.size is not None and info.size > MAX_OBJECT_BYTES:
             summary.skipped_large += 1
             continue
@@ -345,6 +368,30 @@ def fetch_captures(
     return summary
 
 
+def _newest_per_suite(
+    listed: Sequence[tuple[ObjectInfo, str, Path]], *, since: datetime | None, limit: int
+) -> frozenset[str]:
+    """The keys of each suite's newest ``limit`` captures inside ``since``.
+
+    Newest by ``last_modified``; an unknown one counts as oldest, and ties go by key, so every
+    machine listing the same store picks the same captures.
+    """
+
+    def newest_first(info: ObjectInfo) -> tuple[bool, float, str]:
+        modified = info.last_modified
+        return (modified is None, -modified.timestamp() if modified else 0.0, info.key)
+
+    by_suite: dict[str, list[ObjectInfo]] = {}
+    for info, suite_seg, _ in listed:
+        modified = info.last_modified
+        if since is not None and modified is not None and modified < since:
+            continue
+        by_suite.setdefault(suite_seg, []).append(info)
+    return frozenset(
+        info.key for infos in by_suite.values() for info in sorted(infos, key=newest_first)[:limit]
+    )
+
+
 def _absolute(path: Path) -> Path:
     """``path`` made absolute and normalized lexically (``..`` collapsed, drive applied).
 
diff --git a/src/evalshift_cli/captures/selection.py b/src/evalshift_cli/captures/selection.py
index 8e483b1..d960a3f 100644
--- a/src/evalshift_cli/captures/selection.py
+++ b/src/evalshift_cli/captures/selection.py
@@ -48,19 +48,28 @@ class CandidateGroup:
         key: Stable identity used for the deterministic order.
         capture_ids: Its turns' case names (the capture ids), in turn order.
         route: The union route of its turns.
-        content_keys: Each turn's ``example_content_key``.
+        content_keys: Each turn's ``example_content_key``, in turn order.
     """
 
     key: str
     capture_ids: tuple[str, ...]
     route: str
-    content_keys: frozenset[str]
+    content_keys: tuple[str, ...]
 
     @property
     def size(self) -> int:
         """How many cases the group adds."""
         return len(self.capture_ids)
 
+    def prefix(self, turns: int) -> CandidateGroup:
+        """The group's first ``turns`` turns: a conversation cut short, its history intact."""
+        return CandidateGroup(
+            key=self.key,
+            capture_ids=self.capture_ids[:turns],
+            route=self.route,
+            content_keys=self.content_keys[:turns],
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class SuitePlan:
@@ -102,7 +111,9 @@ def plan_suite(
     1. Reviewed cases are pinned; every other case is fresh or stale.
     2. Free slots (``max_cases`` minus pinned and fresh) are filled one group at a time from
        the route least represented so far (ties by route name), each route's groups in
-       ``sha256(key)`` order. A group is added whole, or skipped when it does not fit.
+       ``sha256(key)`` order. A group is added whole, or skipped when it does not fit. Once no
+       whole group fits, the slots left take the leading turns of the next group in that
+       order, so a long conversation cannot leave a suite nearly empty.
     3. A stale case whose content an added turn replays is retired outright, so no duplicate
        content survives. Other stale cases leave only while the suite is over ``max_cases``:
        recorded-on-another-model first, then oldest.
@@ -132,6 +143,11 @@ def plan_suite(
         added.append(group)
         counts[route] += group.size
         slots -= group.size
+    waiting = [r for r, queue in queues.items() if queue]
+    if slots > 0 and waiting:
+        route = min(waiting, key=lambda r: (counts[r], r))
+        added.append(queues[route].pop(0).prefix(slots))
+        counts[route] += slots
 
     retire: list[tuple[str, RetireReason]] = []
     replayed = frozenset(key for group in added for key in group.content_keys)
diff --git a/src/evalshift_cli/cli/commands/capture.py b/src/evalshift_cli/cli/commands/capture.py
index 97fa92b..3551dca 100644
--- a/src/evalshift_cli/cli/commands/capture.py
+++ b/src/evalshift_cli/cli/commands/capture.py
@@ -22,7 +22,7 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Annotated, Any
+from typing import Annotated, Any, Final
 
 import typer
 from rich.console import Console
@@ -367,7 +367,7 @@ def _candidate_groups(pending: Sequence[PromotedCase]) -> list[CandidateGroup]:
             key=key,
             capture_ids=tuple(c.name for c in cases),
             route=examples_route(c.example for c in cases),
-            content_keys=frozenset(example_content_key(c.example) for c in cases),
+            content_keys=tuple(example_content_key(c.example) for c in cases),
         )
         for key, cases in turns.items()
     ]
@@ -378,9 +378,12 @@ def _reason_counts(counts: Counter[str], *, max_age: timedelta | None) -> str:
     labels = {
         "age": f"older than {format_duration(max_age)}" if max_age is not None else "too old",
         "model": "on another model",
+        "history": "missing earlier turns",
         "over_cap": "over max_cases",
     }
-    return ", ".join(f"{counts[k]} {labels[k]}" for k in ("age", "model", "over_cap") if counts[k])
+    return ", ".join(
+        f"{counts[k]} {labels[k]}" for k in ("age", "model", "history", "over_cap") if counts[k]
+    )
 
 
 def _plan_line(
@@ -480,6 +483,20 @@ def _records_own_history(envelope: CaptureEnvelope) -> bool:
     return recovered is not None and recovered.history is not None
 
 
+def _misses_earlier_turns(envelope: CaptureEnvelope, turns: Mapping[str, set[int]]) -> bool:
+    """Whether the turn's history would be rebuilt from earlier turns that are not all local.
+
+    ``turns`` maps each conversation to the turn indices (0-based) on disk. A turn that recorded
+    its own messages list needs none of them; a turn without an index can't be checked.
+    """
+    if envelope.conversation_id is None or envelope.turn_index is None:
+        return False
+    if _records_own_history(envelope):
+        return False
+    present = turns.get(envelope.conversation_id, set())
+    return any(i not in present for i in range(envelope.turn_index))
+
+
 def _warn_unrepairable(
     selection: RepairSelection, records: list[CaptureRecord], *, console: Console
 ) -> None:
@@ -561,6 +578,11 @@ def _warn_store_skipped(config_path: Path, reason: str, *, console: Console) ->
     )
 
 
+FETCH_BUDGET_PER_CASE: Final = 3
+"""With ``captures.max_cases`` set, sync's fetch takes each suite's newest ``3 x max_cases``
+captures: spare enough for the planner to balance tool routes, and never the whole bucket."""
+
+
 @dataclass(frozen=True, slots=True)
 class _SyncLimits:
     """``captures.max_age`` / ``captures.max_cases`` and ``defaults.source_model``, for sync.
@@ -577,6 +599,17 @@ class _SyncLimits:
     def active(self) -> bool:
         return self.max_age is not None or self.max_cases is not None
 
+    @property
+    def fetch_budget(self) -> int | None:
+        """How many of each suite's newest captures the fetch takes; ``None``: all of them."""
+        return FETCH_BUDGET_PER_CASE * self.max_cases if self.max_cases is not None else None
+
+    def fetch_window(self, now: datetime) -> tuple[datetime, str] | None:
+        """The cutoff ``max_age`` puts on the fetch, and how its summary names it."""
+        if self.max_age is None:
+            return None
+        return now - self.max_age, f"captures.max_age ({format_duration(self.max_age)})"
+
 
 def _sync_limits(config_path: Path, *, console: Console, warn: bool) -> _SyncLimits:
     """Read the limits from ``config_path``; none at all when it is missing or unreadable.
@@ -625,11 +658,13 @@ def _fetch_into_base(
     console: Console,
     force_ids: frozenset[str] = frozenset(),
     since_label: str = "--since",
+    per_suite_limit: int | None = None,
 ) -> FetchSummary:
     """Mirror ``store`` into ``base`` and print the one-line summary; exit 1 on failure.
 
     ``force_ids`` are fetched even when promoted or older than ``since`` -- the captures
-    ``capture sync --repair`` needs to rebuild promoted cases.
+    ``capture sync --repair`` needs to rebuild promoted cases. ``per_suite_limit`` caps each
+    suite at its newest that-many captures (``fetch_captures``).
     """
     skip_ids = _promoted_capture_ids_or_warn(base=base, console=console)
     try:
@@ -641,6 +676,7 @@ def _fetch_into_base(
             skip_ids=skip_ids,
             force_ids=force_ids,
             since_label=since_label,
+            per_suite_limit=per_suite_limit,
         )
     except RemoteStoreError as exc:
         console.print(exc.format_rich())
@@ -661,11 +697,12 @@ def _maybe_fetch(
     console: Console,
     force_ids: frozenset[str] = frozenset(),
     window: tuple[datetime, str] | None = None,
+    per_suite_limit: int | None = None,
 ) -> None:
     """The automatic fetch step ``sync`` and ``list`` run before reading the local mirror.
 
     ``window`` is the cutoff ``captures.max_age`` implies, and how to name it; an explicit
-    ``--since`` wins.
+    ``--since`` wins. ``per_suite_limit`` is the budget ``captures.max_cases`` implies.
     """
     since_dt = _resolve_since(since)
     since_label = "--since"
@@ -688,6 +725,7 @@ def _maybe_fetch(
         since_label=since_label,
         console=console,
         force_ids=force_ids,
+        per_suite_limit=per_suite_limit,
     )
 
 
@@ -705,9 +743,15 @@ def capture_list(
     offline: _OfflineOption = False,
     base: _BaseOption = None,
 ) -> None:
-    """List captures recorded by the SDK."""
+    """List captures recorded by the SDK.
+
+    Fetches from captures.store first, as capture sync does: only captures within
+    captures.max_age and, with captures.max_cases set, each suite's newest 3x
+    max_cases.
+    """
     # With --json, stdout carries only the JSON: the fetch line and every warning go to stderr.
     console = Console(stderr=as_json)
+    limits = _sync_limits(config_path, console=console, warn=False)
     _maybe_fetch(
         config_path=config_path,
         offline=offline,
@@ -715,6 +759,8 @@ def capture_list(
         suite=suite,
         base=base,
         console=console,
+        window=limits.fetch_window(_now()),
+        per_suite_limit=limits.fetch_budget,
     )
     records = iter_captures(suite=suite, base=base, on_error=_warn_unreadable(console))
     promoted = _promoted_capture_ids_or_warn(base=base, console=console)
@@ -1354,17 +1400,14 @@ def capture_sync(
     each suite keeps at most that many cases, swapping stale ones (older than
     captures.max_age, or recorded on another model than defaults.source_model)
     for fresh captures; reviewed cases are never retired. captures.max_age
-    keeps older captures out of a suite.
+    keeps older captures out of a suite. The fetch from captures.store takes
+    only captures within captures.max_age and, with captures.max_cases set,
+    each suite's newest 3x max_cases.
     """
     console = Console()
     _check_rounds(rounds)
     limits = _sync_limits(config_path, console=console, warn=True)
     now = _now()
-    window = (
-        (now - limits.max_age, f"captures.max_age ({format_duration(limits.max_age)})")
-        if limits.max_age is not None
-        else None
-    )
     selection = _repair_selection(base=base, suite=suite, console=console)
     _maybe_fetch(
         config_path=config_path,
@@ -1374,7 +1417,8 @@ def capture_sync(
         base=base,
         console=console,
         force_ids=selection.refetch_ids if repair else frozenset(),
-        window=window,
+        window=limits.fetch_window(now),
+        per_suite_limit=limits.fetch_budget,
     )
     if not repair:
         for line in stringified_warnings(selection):
@@ -1511,8 +1555,13 @@ def capture_sync(
         replaceable_from = {c.from_capture for path, c in kept if path in replaceable}
         ineligible: Counter[str] = Counter()
         pending: list[PromotedCase] = []
+        pending_warnings: dict[str, list[str]] = {}
         built_pairs = build_conversation_examples(suite_records, opts, base=base)
         local_ids = {r.envelope.capture_id for r in suite_records}
+        local_turns: dict[str, set[int]] = {}
+        for r in suite_records:
+            if r.envelope.conversation_id is not None and r.envelope.turn_index is not None:
+                local_turns.setdefault(r.envelope.conversation_id, set()).add(r.envelope.turn_index)
         # Re-exercising an agent on the same data records duplicate captures.
         # Promoting both would duplicate the example, double n, and inflate
         # every downstream p-value/effect size. Dedup keys on the *built
@@ -1643,6 +1692,11 @@ def capture_sync(
                 if reason is not None:
                     ineligible[reason] += 1
                     continue
+                # Earlier turns outside the fetch window or budget, or promoted elsewhere and so
+                # never fetched: the history rebuilt without them would be wrong.
+                if _misses_earlier_turns(envelope, local_turns):
+                    ineligible["history"] += 1
+                    continue
             if limits.max_cases is not None and envelope.capture_id in kept_from and not force:
                 skipped_existing += 1
                 continue
@@ -1658,8 +1712,6 @@ def capture_sync(
                 # A replaceable stale case refreshed under --force still leaves its content
                 # to a fresh twin, whichever capture id sorts first.
                 seen_examples.setdefault(content_key, envelope.capture_id)
-            for warning in built.warnings:
-                console.print(f"[yellow]⚠[/yellow] {envelope.capture_id}: {warning}")
             captured_at, source_model = capture_provenance(envelope)
             case = PromotedCase(
                 name=envelope.capture_id,
@@ -1684,8 +1736,12 @@ def capture_sync(
                 if target in kept_paths or (target.exists() and not force):
                     skipped_existing += 1
                     continue
+                # Its warnings print once it is written: a capture the plan leaves out is quiet.
                 pending.append(case)
+                pending_warnings[case.name] = built.warnings
                 continue
+            for warning in built.warnings:
+                console.print(f"[yellow]⚠[/yellow] {envelope.capture_id}: {warning}")
             try:
                 write_promoted_case(case, base=base, force=force)
                 promoted += 1
@@ -1696,6 +1752,16 @@ def capture_sync(
         if repaired_here:
             repaired += len(repaired_here)
             _warn_repaired_duplicates(suite_name, repaired_here, base=base, console=console)
+        if ineligible["history"]:
+            console.print(
+                f"[yellow]⚠[/yellow] {escape(suite_name)}: {ineligible['history']} turn(s) not "
+                "promoted: earlier turns of their conversation are not on disk (outside "
+                "captures.max_age or the fetch budget, promoted on another machine or removed by "
+                "capture clean, sampled out by EVALSHIFT_SAMPLE_RATE, or a turn_index that does "
+                "not start at 0). Pass the full messages list to capture.model_call so each turn "
+                "carries its own history.",
+                soft_wrap=True,
+            )
         suite_line: str | None = None
         if limits.max_cases is not None:
             # Planned by file stem, which is unique in the directory, so a retirement deletes
@@ -1723,6 +1789,8 @@ def capture_sync(
                 except FileExistsError:
                     skipped_existing += 1
                     continue
+                for warning in pending_warnings[case.name]:
+                    console.print(f"[yellow]⚠[/yellow] {case.name}: {warning}")
                 written.append(case)
                 promoted += 1
                 if case.example.generation_config:
diff --git a/tests/unit/test_captures_remote.py b/tests/unit/test_captures_remote.py
index f12ba61..336ead0 100644
--- a/tests/unit/test_captures_remote.py
+++ b/tests/unit/test_captures_remote.py
@@ -397,3 +397,87 @@ def test_skipped_old_names_the_window_it_was_measured_against() -> None:
 
 def test_skipped_old_defaults_to_the_since_flag() -> None:
     assert "1 older than --since" in FetchSummary(store_uri="s3://b/p", skipped_old=1).describe()
+
+
+# --- per-suite budget (captures.max_cases) ---------------------------------------------------
+
+_T = datetime(2026, 10, 19, 12, 0, tzinfo=UTC)
+
+
+def _aged(store: FakeRemoteStore, *hours: tuple[str, float], suite: str = "alpha") -> None:
+    """Stamp each capture's last_modified that many hours before ``_T``."""
+    for cid, h in hours:
+        store.modified[f"captures/{suite}/{cid}.json"] = _T - timedelta(hours=h)
+
+
+def test_per_suite_limit_downloads_each_suites_newest_captures(tmp_path: Path) -> None:
+    store = populated_store("cap_new", "cap_mid", "cap_old")
+    _aged(store, ("cap_new", 1), ("cap_mid", 2), ("cap_old", 3))
+    store.objects["captures/beta/cap_b.json"] = capture_payload("cap_b", "beta")
+
+    summary = fetch_captures(store, base=tmp_path, per_suite_limit=2)
+
+    assert _local_captures(tmp_path) == {
+        "alpha/cap_new.json",
+        "alpha/cap_mid.json",
+        "beta/cap_b.json",
+    }
+    assert summary.skipped_budget == 1
+
+
+def test_per_suite_limit_counts_local_and_promoted_captures_so_a_rerun_fetches_nothing(
+    tmp_path: Path,
+) -> None:
+    store = populated_store("cap_new", "cap_mid", "cap_old")
+    _aged(store, ("cap_new", 1), ("cap_mid", 2), ("cap_old", 3))
+    fetch_captures(store, base=tmp_path, skip_ids={"cap_new"}, per_suite_limit=2)
+    assert _local_captures(tmp_path) == {"alpha/cap_mid.json"}
+    store.gets.clear()
+
+    summary = fetch_captures(store, base=tmp_path, skip_ids={"cap_new"}, per_suite_limit=2)
+
+    assert store.gets == []
+    assert summary.skipped_budget == 1
+
+
+def test_captures_outside_the_window_take_no_budget(tmp_path: Path) -> None:
+    store = populated_store("cap_new", "cap_mid", "cap_old")
+    _aged(store, ("cap_new", 1), ("cap_mid", 2), ("cap_old", 48))
+
+    summary = fetch_captures(store, base=tmp_path, since=_T - timedelta(days=1), per_suite_limit=2)
+
+    assert summary.captures == 2
+    assert summary.skipped_old == 1
+    assert summary.skipped_budget == 0
+
+
+def test_forced_ids_are_fetched_beyond_the_budget(tmp_path: Path) -> None:
+    store = populated_store("cap_new", "cap_old")
+    _aged(store, ("cap_new", 1), ("cap_old", 2))
+
+    fetch_captures(store, base=tmp_path, per_suite_limit=1, force_ids={"cap_old"})
+
+    assert _local_captures(tmp_path) == {"alpha/cap_new.json", "alpha/cap_old.json"}
+
+
+def test_an_unknown_last_modified_counts_as_oldest_for_the_budget(tmp_path: Path) -> None:
+    store = populated_store("cap_known", "cap_unknown")
+    _aged(store, ("cap_known", 100))
+    store.modified["captures/alpha/cap_unknown.json"] = None
+
+    fetch_captures(store, base=tmp_path, per_suite_limit=1)
+
+    assert _local_captures(tmp_path) == {"alpha/cap_known.json"}
+
+
+def test_equal_last_modified_breaks_ties_by_key(tmp_path: Path) -> None:
+    store = populated_store("cap_b", "cap_a", "cap_c")
+
+    fetch_captures(store, base=tmp_path, per_suite_limit=2)
+
+    assert _local_captures(tmp_path) == {"alpha/cap_a.json", "alpha/cap_b.json"}
+
+
+def test_skipped_budget_names_the_limit() -> None:
+    summary = FetchSummary(store_uri="s3://b/p", skipped_budget=4, per_suite_limit=600)
+    assert "4 beyond each suite's newest 600" in summary.describe()
diff --git a/tests/unit/test_captures_selection.py b/tests/unit/test_captures_selection.py
index 38191a9..3038971 100644
--- a/tests/unit/test_captures_selection.py
+++ b/tests/unit/test_captures_selection.py
@@ -32,14 +32,14 @@ def _kept(
 
 
 def _group(
-    key: str, *, route: str = "a", size: int = 1, content: set[str] | None = None
+    key: str, *, route: str = "a", size: int = 1, content: list[str] | None = None
 ) -> CandidateGroup:
     ids = tuple(f"{key}-t{i}" for i in range(size)) if size > 1 else (key,)
     return CandidateGroup(
         key=key,
         capture_ids=ids,
         route=route,
-        content_keys=frozenset(content or {f"k-{i}" for i in ids}),
+        content_keys=tuple(content or [f"k-{i}" for i in ids]),
     )
 
 
@@ -65,10 +65,33 @@ def test_the_least_represented_route_fills_first() -> None:
     assert [g.route for g in plan.add] == ["b", "b", "a"]
 
 
-def test_a_group_is_added_whole_or_not_at_all() -> None:
-    plan = plan_suite([], [_group("conv", size=3), _group("solo", route="b")], max_cases=2)
-    assert _added(plan) == ["solo"]
-    assert plan.added_cases == 1
+def test_a_group_that_fits_is_added_whole() -> None:
+    plan = plan_suite([], [_group("conv", size=3), _group("solo", route="b")], max_cases=4)
+    assert [g.capture_ids for g in plan.add] == [("conv-t0", "conv-t1", "conv-t2"), ("solo",)]
+
+
+def test_slots_no_whole_group_fits_take_a_conversations_leading_turns() -> None:
+    # A lone capture and a 10-turn conversation under max_cases 10: the lone capture leaves
+    # 9 slots, too few for the conversation, so its first 9 turns fill them.
+    plan = plan_suite([], [_group("conv", size=10, route="b"), _group("solo")], max_cases=10)
+    assert [g.capture_ids for g in plan.add] == [
+        ("solo",),
+        tuple(f"conv-t{i}" for i in range(9)),
+    ]
+    assert plan.added_cases == 10
+
+
+def test_a_prefix_is_taken_only_once_no_whole_group_fits() -> None:
+    groups = [_group("conv", size=3), _group("s1", route="b"), _group("s2", route="b")]
+    plan = plan_suite([], groups, max_cases=2)
+    assert sorted(_added(plan)) == ["s1", "s2"]
+
+
+def test_a_prefix_keeps_its_turns_order_and_content_keys() -> None:
+    prefix = _group("conv", size=3).prefix(2)
+    assert prefix.key == "conv"
+    assert prefix.capture_ids == ("conv-t0", "conv-t1")
+    assert prefix.content_keys == ("k-conv-t0", "k-conv-t1")
 
 
 def test_stale_cases_stay_when_nothing_replaces_them() -> None:
@@ -95,7 +118,7 @@ def test_another_model_leaves_before_old_age() -> None:
 
 def test_a_stale_twin_is_replaced_even_under_the_cap() -> None:
     kept = [_kept("twin", stale="age", key="k-q")]
-    plan = plan_suite(kept, [_group("new", content={"k-q"})], max_cases=10)
+    plan = plan_suite(kept, [_group("new", content=["k-q"])], max_cases=10)
     assert _added(plan) == ["new"]
     assert plan.retire == (("twin", "age"),)
 
diff --git a/tests/unit/test_cli_capture_limits.py b/tests/unit/test_cli_capture_limits.py
index b383a5a..fb8fe68 100644
--- a/tests/unit/test_cli_capture_limits.py
+++ b/tests/unit/test_cli_capture_limits.py
@@ -4,6 +4,7 @@
 
 import json
 import re
+from dataclasses import replace
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -36,7 +37,7 @@ def _put(
     suite: str = "alpha",
     created_at: str = WEEK_1,
     model_id: str = "claude-sonnet-5-20260301",
-    text: str | None = None,
+    text: Any = None,
     conversation_id: str | None = None,
     turn_index: int | None = None,
 ) -> None:
@@ -273,7 +274,7 @@ def test_suite_flag_leaves_other_suites_alone(
     assert len(_cases(tmp_path, "beta")) == 5
 
 
-def test_a_conversation_enters_whole_or_not_at_all(
+def test_a_conversation_too_long_for_the_slots_left_enters_as_its_leading_turns(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     for turn in range(3):
@@ -281,10 +282,128 @@ def test_a_conversation_enters_whole_or_not_at_all(
     _put(tmp_path, "solo")
     _config(tmp_path / "evalshift.yaml", max_cases=2)
     _at(monkeypatch, 12)
-
     _sync(tmp_path)
+    suite_dir = tmp_path / "suites" / "alpha"
+    before = {p.name: p.read_bytes() for p in suite_dir.iterdir()}
+
+    result = _sync(tmp_path)
+
+    assert _cases(tmp_path) == {"solo", "conv_t0"}
+    assert {p.name: p.read_bytes() for p in suite_dir.iterdir()} == before
+    assert "alpha: 2 cases — kept 2, retired 0, added 0 of 2 eligible" in _flat(result)
+
+
+def test_a_turn_whose_earlier_turns_are_not_local_is_not_eligible(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Turns 0-1 were promoted elsewhere, or fell outside the fetch budget: turns 2-3 recorded
+    # a bare string, so their history could only be rebuilt from turns that are not here.
+    for turn in (2, 3):
+        _put(tmp_path, f"conv_t{turn}", conversation_id="c1", turn_index=turn)
+    _put(tmp_path, "solo")
+    _config(tmp_path / "evalshift.yaml", max_cases=5)
+    _at(monkeypatch, 12)
+
+    result = _sync(tmp_path)
 
     assert _cases(tmp_path) == {"solo"}
+    flat = _flat(result)
+    assert "not eligible: 2 missing earlier turns" in flat
+    assert "alpha: 2 turn(s) not promoted: earlier turns of their conversation" in flat
+    assert "a turn_index that does not start at 0" in flat
+
+
+def test_max_age_alone_reports_turns_missing_earlier_turns(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _put(tmp_path, "conv_t1", conversation_id="c1", turn_index=1)
+    _config(tmp_path / "evalshift.yaml", max_cases=None)
+    _at(monkeypatch, 12)
+
+    result = _sync(tmp_path)
+
+    assert _cases(tmp_path) == set()
+    assert "alpha: not eligible: 1 missing earlier turns" in _flat(result)
+
+
+def test_a_turn_without_an_index_is_never_missing_earlier_turns(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _put(tmp_path, "conv_a", conversation_id="c1")
+    _put(tmp_path, "conv_b", conversation_id="c1")
+    _config(tmp_path / "evalshift.yaml", max_cases=5)
+    _at(monkeypatch, 12)
+
+    _sync(tmp_path)
+
+    assert _cases(tmp_path) == {"conv_a", "conv_b"}
+
+
+def test_an_earlier_turn_that_is_local_but_not_eligible_still_counts_as_present(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _put(tmp_path, "conv_t0", conversation_id="c1", turn_index=0, model_id="gpt-4o")
+    _put(tmp_path, "conv_t1", conversation_id="c1", turn_index=1)
+    _config(tmp_path / "evalshift.yaml", max_cases=5)
+    _at(monkeypatch, 12)
+
+    result = _sync(tmp_path)
+
+    assert _cases(tmp_path) == {"conv_t1"}
+    assert "not eligible: 1 on another model" in _flat(result)
+
+
+def test_a_turn_that_recorded_its_own_history_needs_no_earlier_turns(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    messages = [
+        {"role": "user", "content": "convert 10 EUR"},
+        {"role": "assistant", "content": "10 EUR is 10.87 USD"},
+        {"role": "user", "content": "and to GBP?"},
+    ]
+    _put(tmp_path, "conv_t1", conversation_id="c1", turn_index=1, text=messages)
+    _config(tmp_path / "evalshift.yaml", max_cases=5)
+    _at(monkeypatch, 12)
+
+    _sync(tmp_path)
+
+    assert _cases(tmp_path) == {"conv_t1"}
+
+
+def test_without_limits_a_turn_missing_earlier_turns_still_promotes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _put(tmp_path, "conv_t2", conversation_id="c1", turn_index=2)
+    _config(tmp_path / "evalshift.yaml", max_age=None, max_cases=None)
+    _at(monkeypatch, 12)
+
+    _sync(tmp_path)
+
+    assert _cases(tmp_path) == {"conv_t2"}
+
+
+def test_build_warnings_name_only_the_captures_that_became_cases(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    real = capture_module.build_conversation_examples
+
+    def warn_every_capture(*args: Any, **kwargs: Any) -> Any:
+        return [
+            (record, replace(built, warnings=[*built.warnings, "WARNED"]))
+            for record, built in real(*args, **kwargs)
+        ]
+
+    monkeypatch.setattr(capture_module, "build_conversation_examples", warn_every_capture)
+    for i in range(5):
+        _put(tmp_path, f"cap_{i}")
+    _config(tmp_path / "evalshift.yaml", max_cases=2)
+    _at(monkeypatch, 12)
+
+    result = _sync(tmp_path)
+
+    warned = set(re.findall(r"(cap_\d): WARNED", result.stdout))
+    assert warned == _cases(tmp_path)
+    assert len(warned) == 2
 
 
 def test_the_same_question_captured_fresh_replaces_its_stale_case(
diff --git a/tests/unit/test_cli_capture_remote.py b/tests/unit/test_cli_capture_remote.py
index 8a97af1..7895ac0 100644
--- a/tests/unit/test_cli_capture_remote.py
+++ b/tests/unit/test_cli_capture_remote.py
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
@@ -15,7 +15,7 @@
 from evalshift_cli.cli.commands.init import CAPTURES_SCAFFOLD, render_minimal_config
 from evalshift_cli.cli.main import app
 from evalshift_cli.suite.loader import load_jsonl
-from tests.unit.fake_remote_store import FakeRemoteStore, populated_store
+from tests.unit.fake_remote_store import FakeRemoteStore, capture_payload, populated_store
 
 runner = CliRunner()
 
@@ -384,6 +384,86 @@ def test_an_explicit_since_overrides_max_age(
     assert "fetched 2 capture(s)" in result.stdout
 
 
+def _with_max_cases(config: Path, max_cases: int) -> Path:
+    config.write_text(
+        config.read_text(encoding="utf-8") + f"  max_cases: {max_cases}\n", encoding="utf-8"
+    )
+    return config
+
+
+def _populate(remote: FakeRemoteStore, count: int) -> None:
+    """``count`` captures in suite alpha, cap_0 the newest, one hour apart."""
+    remote.objects = {k: v for k, v in remote.objects.items() if not k.startswith("captures/")}
+    for i in range(count):
+        key = f"captures/alpha/cap_{i}.json"
+        remote.objects[key] = capture_payload(f"cap_{i}", model_input=f"question {i}?")
+        remote.modified[key] = _NOW - timedelta(hours=i + 1)
+
+
+def test_sync_downloads_at_most_three_times_max_cases_newest_first(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, remote: FakeRemoteStore
+) -> None:
+    _populate(remote, 8)
+    config = _with_max_cases(_write_config(tmp_path / "evalshift.yaml"), 2)
+    monkeypatch.setattr(capture_module, "_now", lambda: _NOW)
+
+    result = _invoke(["sync", "--config", str(config)], tmp_path)
+
+    assert result.exit_code == 0, result.stdout
+    flat = " ".join(result.stdout.split())
+    assert "fetched 6 capture(s)" in flat
+    assert "2 beyond each suite's newest 6" in flat
+    local = {p.stem for p in (tmp_path / "captures" / "alpha").glob("*.json")}
+    assert local == {f"cap_{i}" for i in range(6)}
+
+
+def test_since_replaces_the_age_cutoff_but_not_the_budget(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, remote: FakeRemoteStore
+) -> None:
+    _populate(remote, 8)
+    remote.modified["captures/alpha/cap_7.json"] = datetime(2026, 5, 1, tzinfo=UTC)
+    config = _with_max_cases(_with_max_age(_write_config(tmp_path / "evalshift.yaml")), 2)
+    monkeypatch.setattr(capture_module, "_now", lambda: _NOW)
+
+    result = _invoke(["sync", "--since", "2026-04-01", "--config", str(config)], tmp_path)
+
+    assert result.exit_code == 0, result.stdout
+    flat = " ".join(result.stdout.split())
+    assert "fetched 6 capture(s)" in flat
+    assert "older than" not in flat
+    assert "2 beyond each suite's newest 6" in flat
+
+
+def test_sync_without_max_cases_downloads_every_capture(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, remote: FakeRemoteStore
+) -> None:
+    _populate(remote, 8)
+    config = _write_config(tmp_path / "evalshift.yaml")
+    monkeypatch.setattr(capture_module, "_now", lambda: _NOW)
+
+    result = _invoke(["sync", "--config", str(config)], tmp_path)
+
+    assert result.exit_code == 0, result.stdout
+    assert "fetched 8 capture(s)" in result.stdout
+
+
+def test_list_fetches_what_sync_would(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, remote: FakeRemoteStore
+) -> None:
+    _populate(remote, 8)
+    remote.modified["captures/alpha/cap_7.json"] = datetime(2026, 5, 1, tzinfo=UTC)
+    config = _with_max_cases(_with_max_age(_write_config(tmp_path / "evalshift.yaml")), 2)
+    monkeypatch.setattr(capture_module, "_now", lambda: _NOW)
+
+    result = _invoke(["list", "--config", str(config)], tmp_path)
+
+    assert result.exit_code == 0, result.stdout
+    flat = " ".join(result.stdout.split())
+    assert "fetched 6 capture(s)" in flat
+    assert "1 older than captures.max_age (14d)" in flat
+    assert "1 beyond each suite's newest 6" in flat
+
+
 def test_max_age_without_a_store_is_not_a_fetch_error(tmp_path: Path) -> None:
     config = _write_config(tmp_path / "evalshift.yaml", store=None)
     config.write_text(