feat: Implement well-known typed fields for verdicts in output models

- Introduced `_well_known_specs` function to define typed fields for `winner` and `scores` in agent output.
- Updated `build_output_model` to handle new field types, allowing `winner` to be optional and `scores` to default to an empty map.
- Enhanced `json_instruction` to provide specific hints for well-known fields in the output format.
- Modified `FishbowlSession` to derive `winner_kind` and `winning_models` based on verdicts, supporting team-based outcomes.
- Added validation tests for new competition configurations and verdict handling, ensuring proper behavior for both agent and team winners.
- Created comprehensive tests for the new verdict validation logic, including re-ask scenarios and score normalization.

config/agents/mystery-judge.yaml

@@ -4,7 +4,10 @@ persona: >
   You are the Mystery Judge. After reviewing the clues and debate, declare the most
   likely explanation in one confident sentence. Start with 'Verdict:'. Choose the
   most interesting, specific answer the evidence supports.
-  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip),
+  your `winner` (the exact on-stage name of the mind whose hypothesis your verdict
+  endorses), and a `scores` map from each mind's on-stage name to a 0–10 rating of how
+  much their contribution supported the answer.
 subscribes_to: []
 may_emit:
   - judge.verdict
@@ -16,6 +19,6 @@ memory:
   use_salience: true
   salience_top_k: 8
 tools: []
-output_extra_fields: [mood]
+output_extra_fields: [mood, winner, scores]
 hue: 320
 archetype: the mystery judge

config/agents/spy-host.yaml

@@ -8,7 +8,9 @@ persona: >
   dissect" or "let the clues be cast". Weigh who sat oddly: whose clue fit a different drink,
   who slipped a tell, who over-covered. Begin exactly with "Verdict:" then name that one mind
   and the single tell that gave them away, in one or two sentences. This is the final word.
-  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip),
+  your `winner` (the exact on-stage name of the mind you accuse), and a `scores` map from
+  each mind's on-stage name to a 0–10 suspicion rating.
 subscribes_to: []
 may_emit:
   - judge.verdict
@@ -22,6 +24,6 @@ memory:
   use_salience: true
   salience_top_k: 8
 tools: []
-output_extra_fields: [mood]
+output_extra_fields: [mood, winner, scores]
 hue: 300
 archetype: the host

config/scenarios/mystery-roots.yaml

@@ -14,6 +14,10 @@ cast:
   - hypothesis-former
   - devils-advocate
   - mystery-judge
+# A judged scenario (ADR-0029): no ground truth, so the judge's pick *is* the result —
+# its `winner` field names the mind whose hypothesis the verdict endorses.
+competition:
+  kind: judged
 governor:
   max_turns: 60
   max_calls_per_turn: 16

config/scenarios/the-steeped.yaml

@@ -25,6 +25,18 @@ cast:
   - spy-nil
   - spy-ovo
   - spy-host
+# A versus contest with ground truth (ADR-0029): spy-nil holds the near-twin word.
+# The spy-host's verdict accuses a mind; the handler scores that accusation against
+# these teams in code — the herd wins iff the accused really is the spy.
+competition:
+  kind: versus
+  teams:
+    spy:
+      - spy-nil
+    herd:
+      - spy-cara
+      - spy-bex
+      - spy-ovo
 governor:
   max_turns: 2000
   max_calls_per_turn: 16

docs/adr/0029-structured-verdicts.md

@@ -0,0 +1,225 @@
+# ADR-0029: Structured Verdicts — Winners as Data, Not Prose
+
+## Status
+
+Proposed
+
+## Context
+
+A `judge.verdict` today is prose with a `mood`. The run lifecycle (ADR-0026) already
+*wants* a machine-readable winner — `FishbowlSession.finalize` reads
+`verdict.payload.get("winner")` and `RunSummary` carries `winner` / `winning_model` —
+but nothing reliably populates that key: judges emit `{kind, text, mood}` and the
+winner lives only in the sentence "Verdict: Bex slipped the tell." A leaderboard, a
+shareable trace with a scoreboard, and any "who won on what model" claim all need the
+winner as data.
+
+Three constraints shape the design:
+
+1. **The structured-output contract is string-typed.** `build_output_model`
+   (`src/core/structured.py`, ADR-0016) makes every `output_extra_fields` entry a
+   *required `str`*. A winner is an optional cast name; scores are a `dict[str, float]`.
+   The offline tolerant parser already passes arbitrary JSON keys through untouched,
+   so only the typed live model and the prompt instruction are string-bound.
+2. **Agents never see scenario config.** `Registry.build_scenario` resolves the cast
+   *before* constructing the `Scenario`; a judge's handler has no view of which
+   scenario it serves, so it cannot know the cast to validate against or the team map
+   to attribute with — unless the registry injects that context (the same seam as
+   `agent.manifest = manifest`).
+3. **Some scenarios have ground truth, some don't, some have neither.** In The Steeped
+   the engine *knows* the spy (`spy-nil`) — the judge's accusation can be scored by
+   code. In Mystery Roots there is no truth; the judge's pick *is* the answer. In
+   Thousand Token Wood nothing wins. One mechanism must serve all three without
+   hard-coding scenario names in the engine.
+
+Prize relevance: a code-stamped scoreboard over an LLM verdict is the cleanest
+"AI is load-bearing for judgment, code is load-bearing for bookkeeping" story
+(Best Agent, Best Demo), and the enriched trace strengthens the Sharing-is-Caring
+export (ADR-0026).
+
+## Decision
+
+Make the winner a first-class, validated payload field, derived by the layer that
+actually knows it: the LLM where judgment is the product, the handler where ground
+truth exists, and never at all where the scenario declares no competition. Five
+additive pieces, `schema_version` stays 1 (ADR-0009).
+
+### 1. Scenario competition contract (`CompetitionConfig`, extends ADR-0011)
+
+`ScenarioConfig` gains an optional block, validated in `src/core/config.py`:
+
+```yaml
+competition:
+  kind: versus | judged | none      # default none; absent block == none
+  teams:                            # versus only
+    spy: [spy-nil]
+    herd: [spy-cara, spy-bex, spy-ovo]
+```
+
+Validation rules (in `CompetitionConfig` + the existing
+`WorldConfig._check_cast_references` validator):
+
+- `teams` is permitted only when `kind: versus`, must be non-empty there, with
+  non-empty, mutually disjoint member lists.
+- Every team member must be in the scenario's `cast`.
+- **No team label may collide with an agent name** — `winner` carries either an agent
+  name or a team label, and this rule is what keeps that union unambiguous.
+
+The Steeped declares `versus` with the teams above; Mystery Roots declares `judged`;
+Thousand Token Wood declares nothing (`none`). `kind: none` scenarios keep full
+sessions/history (ADR-0027) — they simply never produce a winner.
+
+### 2. Well-known typed extra fields (extends ADR-0016)
+
+`output_extra_fields` stays a `list[str]` — no manifest syntax change. Instead,
+`src/core/structured.py` gains a small table of **well-known field types**:
+
+| field    | type                | required |
+|----------|---------------------|----------|
+| `winner` | `str \| None`       | no (default `None`) |
+| `scores` | `dict[str, float]`  | no (default `{}`)   |
+| *other*  | `str`               | yes (unchanged)     |
+
+`build_output_model` consults the table; `json_instruction` renders a typed schema
+hint for known fields (`"winner": "<name or null>"`, `"scores": {"<name>": 0-10}`).
+`winner` and `scores` are not arbitrary scenario fields — they are the verdict
+contract ADR-0026 already names in `run.finished` — so giving them engine-known types
+is the same move as `CORE_EVENT_KINDS`: open surface, curated core. Back-compat is
+total: every existing manifest (`[mood]`, `[thought]`, …) hits the *other* row and
+behaves exactly as before; the offline parser needs no change because it already
+passes non-string values through.
+
+### 3. Winner validation and one re-ask, in the base agent
+
+Validation lives in `ManifestAgent`, not in per-judge handlers — `_resolve_payload`
+owns the model call, so it is the only seam where a re-ask is one extra round-trip
+instead of a re-architecture:
+
+- The registry attaches the scenario's competition context when assembling a cast:
+  `agent.competition = cfg.competition` in `build_scenario`, plus the cast name list
+  and team labels (the *valid winner vocabulary*).
+- A new overridable hook `_validate_payload(parsed) -> str | None` runs after the
+  structured call (and after the offline parse). The base implementation activates
+  only when `role == "judge"`, a competition with `kind != none` is attached, and
+  `winner` is in the agent's extra fields. A present-but-unknown `winner` (not a cast
+  name, not a team label) returns an error string; a missing/`None` winner is *not*
+  an error (the field is optional, and the offline stub never emits it — determinism
+  preserved).
+- On error, `_resolve_payload` re-asks **once**: the same prompt plus a corrective
+  line naming the valid options. Token usage from both calls is *accumulated* into
+  `last_usage` so the governor (ADR-0013) meters the retry. On a second failure the
+  invalid `winner` is dropped and `payload["no_contest"] = true` is stamped — the
+  verdict text still ships (the drama survives), the leaderboard simply gets no row.
+- `scores` is validated but never re-asked (it is garnish, not load-bearing): unknown
+  agent keys are dropped, values clamped to 0–10.
+
+### 4. Ground truth belongs in code (the versus path)
+
+For `kind: versus`, the LLM's `winner` field is its **accusation** (a cast name,
+validated by §3). The scenario's handler — `SpyHost` is the template — then computes
+the scoreboard after `super().act()`:
+
+```
+accused  = payload.pop("winner")            # the judge's pick, kept as payload["accused"]
+correct  = accused in competition.teams["spy"]
+winner   = "herd" if correct else "spy"     # team label, stamped by code
+payload  |= {"accused": accused, "correct": correct, "winner": winner}
+```
+
+The team map comes from `competition.teams`, not from a handler constant — the
+curated `_REVEAL` dict stays what it is (demo content: secrets and reveal drama),
+while *who is on which team* is scenario config. Offline fallback: when the stub's
+verdict carries no `winner`, the handler scans the verdict text for the first cast
+name mentioned and treats it as the accusation; if none is found, `no_contest`. This
+keeps the no-API-key demo producing a full stamped scoreboard, deterministically.
+
+For `kind: judged` (Mystery Roots), no handler is needed: the validated `winner`
+*is* the result — AI is load-bearing for the judgment itself. For `kind: none`,
+judges do not declare `winner` in their manifests at all; `mischief-critic` keeps
+`[mood]` (the Wood's reckoning records what became real — nobody wins it).
+
+### 5. Attribution contract (extends ADR-0026)
+
+`winner` is now an agent name (*judged*) or a team label (*versus*), so the
+`run.finished` payload and `RunSummary` gain two additive keys:
+
+```
+"winner":         str | None    # unchanged — display name for the leaderboard row
+"winner_kind":    "agent" | "team" | None
+"winning_models": list[str]     # model_endpoint of the winner, or of every
+                                # member of the winning team (None entries dropped)
+```
+
+`winning_model` keeps its exact current meaning — a single cast agent's model — and
+is populated only when `winner_kind == "agent"`; it is `None` for team wins (never a
+guess). `FishbowlSession.finalize` resolves the kind by checking the winner against
+the run's cast map first, then the scenario's team labels. A future leaderboard
+renders one row per finished run in a `kind != none` scenario: winner name,
+`correct` badge where present, and the winning model(s). The UI itself is out of
+scope here.
+
+## Consequences
+
+- **The verdict is data and drama at once.** `judge.verdict` carries
+  `winner`/`accused`/`correct`/`scores` machine-readably while `text` stays the
+  spoken ruling; `finalize`'s existing best-effort read starts actually working.
+- **All additive.** No schema bump, no migration; old ledgers, old manifests, and the
+  string-typed extra-field behaviour are untouched. The offline stub emits no
+  `winner`, so the deterministic demo is byte-identical except where the SpyHost
+  text-scan stamps the scoreboard.
+- **One re-ask is bounded cost.** At most one extra model call per verdict, metered
+  by the governor; the failure mode is a missing leaderboard row, never a missing
+  show ending.
+- **The well-known field table is a curated list in engine code.** A scenario cannot
+  invent a new *typed* field from YAML alone — accepted: arbitrary fields remain
+  available as strings, and a handler can always derive structure from them. If a
+  third typed field ever appears, revisit a declarative field-spec syntax (see
+  alternatives).
+- **Validation vocabulary is injected, not discovered.** Agents now carry a small
+  piece of scenario context (`competition`). This is a deliberate, single-attribute
+  seam mirroring `agent.manifest`; it does not give agents the scenario object.
+- **Risk: usage accounting on re-ask.** `last_usage` must sum both calls or the
+  governor undercounts; this is an explicit acceptance criterion, not an afterthought.
+- Prize impact: strengthens Best Agent / Best Demo (code-stamped scoreboard over
+  small-model judgment) and Sharing-is-Caring (self-scoring trace); the future
+  leaderboard it enables feeds Community Choice polish. No track is disqualified;
+  the ≤32B constraint is untouched.
+
+## Alternatives considered
+
+- **Typed field descriptors in manifest YAML** (`output_extra_fields: [{name: scores,
+  type: score_map}]`). Maximally declarative, but adds a config surface and a union
+  schema for exactly two fields the engine already treats as core in ADR-0026.
+  Rejected as not the thinnest slice; the well-known table can grow into this later
+  without breaking `list[str]`.
+- **Keep extra fields string-typed; handlers parse `winner`/`scores` out of strings.**
+  No engine change, but every judge handler re-implements parsing and the live path
+  loses validation-by-construction — the exact regression ADR-0016 exists to prevent.
+- **Validate/re-ask in the conductor or per-handler.** The conductor never re-prompts
+  (it has no prompt), and per-handler re-ask duplicates the retry across
+  spy-host/mystery-judge and inverts the `super().act()` flow. The base-class hook is
+  the only seam that owns both the prompt and the provider.
+- **Let the LLM declare the team winner in versus scenarios.** Simpler wiring, but
+  the model can be *wrong about its own conclusion's consequence* (naming the spy yet
+  declaring the spy won). Where truth exists, code stamps it — this is the
+  load-bearing split the feature exists to demonstrate.
+- **A separate `judge.scored` event kind.** Keeps `judge.verdict` lean, but splits
+  one ruling across two events that every consumer must re-join, and the Fishbowl
+  curtain-fall already keys on the first `judge.verdict`. One enriched event wins.
+- **Reuse `winner` for the accusation and skip `accused`.** Loses the LLM's actual
+  pick once the handler overwrites it; `accused` + `correct` is what makes the trace
+  auditable.
+
+## References
+
+- ADR-0009 (open/additive event kinds — new payload keys, no schema bump)
+- ADR-0011 (declarative validatable config — `CompetitionConfig` joins `ScenarioConfig`)
+- ADR-0016 (validated structured output — `build_output_model` typing extended)
+- ADR-0026 (run lifecycle — `run.finished` winner contract extended with
+  `winner_kind` / `winning_models`)
+- `src/core/structured.py` — well-known field types, typed `json_instruction`
+- `src/core/config.py` — `CompetitionConfig`, cross-validation in `WorldConfig`
+- `src/agents/base.py` — `_validate_payload` hook, single re-ask, usage accumulation
+- `src/agents/handlers.py` — `SpyHost` ground-truth scoreboard
+- `src/core/registry.py` — competition context injection in `build_scenario`
+- `src/ui/fishbowl/session.py`, `src/core/run_index.py` — attribution resolution

docs/adr/0030-gpu-memory-snapshots-cold-start.md

@@ -0,0 +1,121 @@
+# ADR-0030: GPU Memory Snapshots and Keep-Warm for Cold-Start Latency
+
+## Status
+
+Accepted (extends [ADR-0014 *Modal model serving*](0014-modal-model-serving.md),
+[ADR-0019](0019-single-model-catalogue-no-cloud-path.md))
+
+## Context
+
+Every served model scales to zero (`min_containers=0`), so the first request to
+an idle endpoint pays the full cold-start pipeline: container boot → weight
+download/load → engine warmup (CUDA-graph capture, compile cache). On the bigger
+models this is **minutes**, which hurts in two places that matter for the
+hackathon:
+
+- **The live demo.** A judge's first click should not stare at a spinner while a
+  14B model loads — the first 30 seconds are scored.
+- **Iteration speed.** Every healthcheck or engine run against a cold workspace
+  pays the same multi-minute tax per model.
+
+The mitigations we already had are blunt: `min_containers` removes cold starts
+entirely but burns GPU-hours around the clock, and the shared `vllm-cache`
+Volume only amortizes *compilation* — weight load and warmup are still paid by
+every cold container.
+
+Modal's answer to exactly this is **memory snapshots**: checkpoint a booted
+container (CPU state, and with the alpha `enable_gpu_snapshot` flag, GPU state)
+and restore it on later cold starts instead of re-initializing. Modal's
+documented vLLM recipe pairs snapshots with vLLM **sleep mode**: warm the engine,
+offload weights to host RAM (`POST /sleep?level=1`), snapshot, then reload
+weights on restore (`POST /wake_up`). Restores land in seconds.
+
+The catch: the recipe requires a *class-based* lifecycle (`@modal.enter(snap=True)`
+for the snapshotted warmup, `@modal.enter(snap=False)` for post-restore wake),
+while our serving path (ADR-0014) registers plain `@app.function` web servers.
+And GPU snapshots are alpha, with real constraints: single-GPU only, the model's
+vLLM build must support sleep mode, and host RAM must hold the offloaded weights.
+
+## Decision
+
+**1. Snapshots are a per-model catalogue flag, not a global switch.**
+`ModelConfig.gpu_snapshot: bool = False` in `modal/catalogue.py`. When set,
+`service.register_model()` dispatches to a class-based registrar
+(`_register_snapshot_model`) implementing Modal's recipe verbatim:
+
+- `@modal.enter(snap=True)` — start `vllm serve` (with `--enable-sleep-mode`),
+  wait for the port, run three warmup completions so compile/caching work lands
+  *inside* the snapshot, then `POST /sleep?level=1`. `startup_timeout` bounds
+  this whole phase (download + load + warmup + sleep).
+- `@modal.enter(snap=False)` — `POST /wake_up` after every restore (it also
+  runs on the snapshot-creating boot itself, which simply resumes serving).
+- `@modal.web_server(..., label=cfg.endpoint_name)` — a no-op method that
+  exposes the already-running vLLM port. The `label` pins the public URL to
+  `…--<app>-<endpoint_name>.modal.run`, byte-identical to the function path, so
+  clients, the engine catalogue (ADR-0019), and the DNS-label tests are
+  untouched.
+- Image gains `VLLM_SERVER_DEV_MODE=1` (exposes the sleep/wake endpoints) and
+  `TORCHINDUCTOR_COMPILE_THREADS=1` (snapshot-safe compile), both scoped to
+  snapshot models.
+
+Helpers used by the class are **nested closures, not module functions**: the
+class ships via cloudpickle (`serialized=True`), which pickles closures by value
+but module-level functions by reference — and the `service` module doesn't exist
+inside the container. (Verified by round-tripping the pickled class with the
+local modules removed.)
+
+**2. Conservative initial casting.** Snapshots are on for the well-behaved
+single-GPU, native-vLLM models the cast hits hardest — `nemotron-3-nano-4b`
+(tiny), `minicpm-4-1-8b` (fast), `nemotron-cascade-14b` (Judge specialist) — and
+deliberately **off** where the recipe is unproven or impossible:
+
+| Model | Why not |
+| --- | --- |
+| Gemma 4 12B / 26B | Nightly vLLM + Transformers modeling backend; sleep mode unverified on that path. |
+| Nemotron-3-Nano-30B | ~60GB BF16 weights won't fit host RAM during sleep level 1. |
+| MiniCPM-o 4.5 | Omni-modal custom code path; kept conservative like its other knobs. |
+
+Rolling back any model is a one-line `gpu_snapshot=False` — the plain function
+path is untouched by this ADR.
+
+**3. A deploy-time keep-warm switch for demo day.** `MODAL_LLM_KEEP_WARM=N`
+(mirroring the existing `MODAL_LLM_REQUIRE_AUTH` / `MODAL_LLM_JSON_LOGS` idiom)
+raises `min_containers` to N **for profile-bound models only** — the four tiers
+the cast actually runs on. Specialists keep scale-to-zero. This is the
+belt-and-braces for the hours around a live demo; snapshots are the everyday
+path.
+
+## Consequences
+
+- Cold starts on snapshot models drop from minutes to seconds; scale-out under
+  burst gets the same benefit (every new container is a restore, not a boot).
+- The serving layer now has two registration shapes (function vs. class). Both
+  are produced by the same loop from the same `ModelConfig`, and the dispatch is
+  one `if` — but anyone debugging a snapshot model must know the lifecycle runs
+  through `@modal.enter` hooks, not the function body.
+- GPU snapshots are **Modal-alpha**. Known sharp edges we accept and guard:
+  snapshots are invalidated by image changes (safe — they rebuild), restores can
+  fail if Volume files used during snapshot are deleted, and the feature is
+  single-GPU only (all snapshot models are `tensor_parallel_size=1`).
+- The alpha surface is also an **API-stability risk**: the recipe is verified
+  against the pinned Modal SDK (1.4.3 — `App.cls` accepts `serialized` /
+  `enable_memory_snapshot` / `experimental_options`; `modal.web_server` accepts
+  `label`), but `experimental_options={"enable_gpu_snapshot": True}` carries no
+  compatibility promise. Re-verify these kwargs on every SDK bump; the per-model
+  rollback (`gpu_snapshot=False`) restores the untouched plain path.
+- The test suite covers registration, dispatch, and the cloudpickle round-trip,
+  but a snapshot *restore* can only be observed live. Validate each snapshot
+  model with `modal/healthcheck.py` after its first deploy — and again before
+  demo day — rather than trusting the recipe on paper.
+- The snapshot warmup makes three tiny completions at deploy-boot time; with
+  auth enabled they authenticate via the same `VLLM_API_KEY` the secret injects,
+  so `MODAL_LLM_REQUIRE_AUTH=1` keeps working.
+- `MODAL_LLM_KEEP_WARM` left on costs real GPU-hours — it is documented as a
+  demo-window switch, and the default deploy keeps scale-to-zero.
+- First boot per snapshot model is slightly *slower* (warmup + sleep before
+  serving), which is the right trade: it runs once per image/config change, not
+  per cold start.
+- Prize impact: this makes the Modal serving path credibly demo-ready
+  (Modal Awards) and defends the first-30-seconds bar that Best Demo and
+  Community Choice are scored on. The no-API-key deterministic stub is
+  untouched, so the on-stage fallback stays reproducible.

docs/architecture/manifest-spec.md

@@ -143,6 +143,12 @@ scenario shape agent output without engine edits.  The Fishbowl cast uses
 deterministic stub synthesises them offline so the mind-reader works with no API key
 (ADR-0021).
 
+Fields are required strings by default, but two names are **well-known and
+engine-typed** (ADR-0029): `winner` (`str | None`, optional) and `scores`
+(`dict[str, float]`, optional).  Judges in competition scenarios list them —
+`output_extra_fields: [mood, winner, scores]` — to make the verdict machine-readable.
+See [structured-output.md](structured-output.md#well-known-typed-fields).
+
 ### `hue` / `archetype`
 Optional presentation metadata, consumed by the Fishbowl UI presenter and **ignored by
 the engine** (ADR-0021).  `hue` (0–360) colours the agent's mind on stage; `archetype`

docs/architecture/scenario-authoring.md

@@ -375,6 +375,19 @@ techniques:
   the same "decorate the emitted event" move `FortuneTeller` uses for `omen`. Riding the
   real ledger, the reveal scrubs and replays like any other event.
 
+- **Ground truth gets a code-stamped scoreboard.** The scenario declares a
+  [`competition:` block](../schema/scenario-config.md#competition-who-can-win-and-who-decides)
+  (`kind: versus`, ADR-0029) naming the teams — `spy: [spy-nil]` vs the herd. The
+  judge's `winner` field (a well-known typed extra field, validated with one re-ask)
+  is its *accusation*; the `SpyHost` handler then scores it in code: the herd wins
+  iff the accused really is the spy, and the verdict payload gains `accused`,
+  `correct`, and a team-label `winner`. This is the load-bearing split — the model
+  provides the judgment drama, code stamps the bookkeeping where truth exists.
+  (Contrast `mystery-roots`, `kind: judged`: no ground truth, so the model's
+  validated `winner` *is* the result, no handler needed.) Offline, the handler
+  recovers the accusation from the verdict text, so the no-API-key demo still
+  produces a full deterministic scoreboard.
+
 Offline, curated lines + a per-role mood bias in `src/models/provider.py` (`_STUB_*`,
 keyed by agent name) give the bluff a coherent arc with no API key; live, a real small
 model improvises from the personas. Either way the cast never calls each other — they
@@ -390,6 +403,7 @@ only post to the shared log, and the seam shows.
 |---|---|---|
 | a new cast on existing patterns | agent + scenario YAML | none |
 | a hidden-role / secret-info game | secret in each `persona`; reveal via a verdict `handler` | none (handler in `agents/`) |
+| a scenario that produces a *winner* | a `competition:` block; `winner` in the judge's `output_extra_fields` (+ a scoring `handler` if ground truth exists) | none (ADR-0029) |
 | a new event kind | just use it in `may_emit` | none |
 | an agent that calls an existing tool | a `handler` + `tools:` grant | none (handler in `agents/`) |
 | an agent that calls a *new* tool | register it in `builtins.py` + grant it | tool registration only |

docs/architecture/structured-output.md

@@ -140,6 +140,9 @@ They're useful for:
 - Routing decisions (e.g. "if emotion=desperate, escalate to judge")
 - Downstream agent context (the Echo agent could read the emitting agent's "wants")
 
+Most extra fields are required strings, but two names are **well-known and
+engine-typed** — see *Well-known typed fields* below.
+
 ---
 
 ## Testing structured output
@@ -178,6 +181,36 @@ model literally cannot validate with a kind it isn't authorised to emit. The
 function is pure Pydantic — no provider, no network — so it is unit-tested
 directly and importable with the structured-output dependency absent.
 
+### Well-known typed fields
+
+`output_extra_fields` stays a plain `list[str]` — no manifest syntax change — but
+`src/core/structured.py` carries a small table of names the engine knows how to
+type (ADR-0029):
+
+| field    | type                | required            |
+|----------|---------------------|---------------------|
+| `winner` | `str \| None`       | no (default `None`) |
+| `scores` | `dict[str, float]`  | no (default `{}`)   |
+| *other*  | `str`               | yes (unchanged)     |
+
+`winner` and `scores` are not arbitrary scenario fields — they are the verdict
+contract that `run.finished` already names (ADR-0026), so giving them engine-known
+types is the same move as `CORE_EVENT_KINDS`: open surface, curated core. A judge
+manifest lists them like any extra field (`output_extra_fields: [mood, winner,
+scores]` — see `config/agents/mystery-judge.yaml`).
+
+Both halves of the contract honour the table. `build_output_model` makes the typed
+fields optional with defaults, and `json_instruction` renders a typed schema hint
+instead of the generic string slot — `"winner": "<a player's name, or null>"`,
+`"scores": {"<player>": 0-10}` — so a small model knows it may answer `null`.
+Back-compat is total: every existing manifest (`[mood]`, `[thought]`, …) hits the
+*other* row and behaves exactly as before, and the tolerant offline parser already
+passed non-string values through untouched.
+
+What happens to a *validated-but-wrong* `winner` (a name outside the cast) is the
+verdict-validation story — one re-ask, then `no_contest` — documented in
+[events.md](../schema/events.md#verdict-and-run-payloads-adr-0029) and ADR-0029.
+
 ### The structured call
 
 `LiteLLMProvider.complete_structured(role, prompt, response_model)` wraps the

docs/schema/agent-manifest.md

@@ -63,6 +63,10 @@ output_extra_fields: []        # extra payload fields the model is asked for, e.
 - **`handler`** stays `null` for the common case (the generic `ManifestAgent`).
   Set it to a key registered via `@register_handler` for agents that call tools or
   need custom prompt logic; the YAML still supplies all declarative fields.
+- **`output_extra_fields`** entries are required strings, except the well-known
+  typed names `winner` and `scores` (optional, engine-typed — ADR-0029).  A judge in
+  a competition scenario lists them to make its verdict machine-readable; see
+  [structured-output.md](../architecture/structured-output.md#well-known-typed-fields).
 - **`memory.*`** layers are pure views over the ledger — see
   [memory-stack.md](../architecture/memory-stack.md).
 

docs/schema/events.md

@@ -36,7 +36,7 @@ kinds with **zero engine edits** — that is the modularity contract for the sch
 memory importance defaults).  It is a default set, not a gate:
 
 ```
-run.started · world.observed · agent.thought · agent.spoke
+run.started · run.finished · world.observed · agent.thought · agent.spoke
 agent.reflected · judge.verdict · user.injected
 ```
 
@@ -44,6 +44,43 @@ Any other well-formed kind (e.g. `oracle.spoke`, `crier.announced`) is valid and
 if it carries a `text` payload, renders on stage via the generic projection
 fallback.
 
+## Verdict and run payloads (ADR-0029)
+
+In a scenario with a [`competition:` block](scenario-config.md#competition-who-can-win-and-who-decides),
+the winner is data, not just prose.  Two core kinds carry it — all keys additive,
+`schema_version` stays 1 (ADR-0009).
+
+### `judge.verdict`
+
+Alongside the spoken `text` (and any manifest extras like `mood`):
+
+| key | type | when present | meaning |
+|---|---|---|---|
+| `winner` | `str \| None` | `judged` and `versus` | agent name (*judged* — the model's validated pick) or team label (*versus* — stamped by code) |
+| `accused` | `str` | `versus` | the judge's named pick, preserved before code overwrites `winner` — keeps the trace auditable |
+| `correct` | `bool` | `versus` | ground truth: was the accused actually the spy? |
+| `scores` | `dict[str, float]` | when in the judge's `output_extra_fields` | per-agent map, cleaned in code: unknown names dropped, values clamped to 0–10 |
+| `no_contest` | `true` | on failure | the model named an invalid winner and one corrective re-ask didn't fix it; `winner` is dropped, the verdict `text` still ships |
+
+An invalid `winner` (not a cast name, not a team label) triggers **one** re-ask in
+`ManifestAgent` (`src/agents/base.py`), with both calls' token usage summed so the
+governor meters the retry (ADR-0013).  A missing `winner` is never an error — the
+offline stub doesn't emit it, so deterministic demos are unaffected.
+
+### `run.finished`
+
+The attribution contract (ADR-0026, extended by ADR-0029) — mirrored on `RunSummary`:
+
+| key | type | meaning |
+|---|---|---|
+| `winner` | `str \| None` | display name for the leaderboard row — agent name or team label |
+| `winner_kind` | `"agent" \| "team" \| None` | how to read `winner`: checked against the run's cast map first, then team labels |
+| `winning_model` | `str \| None` | unchanged legacy key — a single agent winner's `model_endpoint`; `None` for team wins (never a guess) |
+| `winning_models` | `list[str]` | the winner's endpoint, or every winning-team member's (`None` entries dropped) |
+
+`FishbowlSession.finalize` (`src/ui/fishbowl/session.py`) resolves the kind and
+models when stamping `run.finished`.
+
 ## Evolution Rules
 
 - Add fields instead of renaming fields; keep history immutable.

docs/schema/scenario-config.md

@@ -21,6 +21,8 @@ cast:                          # agent names, resolved via the agent registry
   - devils-advocate
   - mystery-judge
 genesis_text: "A mystery settles over the wood: {seed}"   # '{seed}' substituted
+competition:                   # optional contest contract (ADR-0029); absent == none
+  kind: judged                 # versus | judged | none
 governor:                      # optional per-scenario budget (else defaults)
   max_turns: 2000
   max_calls_per_turn: 16
@@ -38,8 +40,45 @@ governor:                      # optional per-scenario budget (else defaults)
 | `example_seeds` | Seed gallery for the UI dropdown. |
 | `cast` | Agent names that participate.  **Selecting who participates is editing this list.**  Each must exist in `config/agents/`. |
 | `genesis_text` | Template for the opening `world.observed`; `{seed}` is replaced. |
+| `competition` | Optional `CompetitionConfig` — does this scenario produce a winner, and how?  See below. |
 | `governor` | Optional `GovernorConfig`; omit for engine defaults. |
 
+## Competition: who can win, and who decides
+
+A scenario declares whether it produces a winner with the optional `competition:`
+block (`CompetitionConfig`, ADR-0029).  Absent block == `kind: none` — full sessions
+and history, but nobody wins.
+
+```yaml
+competition:
+  kind: versus | judged | none   # default none
+  teams:                         # versus only
+    spy: [spy-nil]
+    herd: [spy-cara, spy-bex, spy-ovo]
+```
+
+The three kinds split *who derives the winner*:
+
+| kind | ground truth? | winner derived by | shipped example |
+|---|---|---|---|
+| `versus` | yes — the team map | **code** — the scenario's handler scores the judge's accusation against `teams` | `the-steeped` |
+| `judged` | no — judgment *is* the result | **the model** — the judge's validated `winner` field | `mystery-roots` |
+| `none` | n/a | nobody — judges don't declare `winner` at all | everything else |
+
+Validation rules (enforced in `CompetitionConfig` and `WorldConfig`,
+`src/core/config.py` — a bad block fails loudly at load, per ADR-0011):
+
+- `teams` is permitted only when `kind: versus`, and is required (non-empty) there.
+- Member lists must be non-empty and **mutually disjoint** — no double agents.
+- Every team member must appear in the scenario's `cast`.
+- **No team label may equal an agent name.**  The `winner` payload key carries either
+  an agent name or a team label; this rule keeps that union unambiguous.
+
+The registry injects the competition context into the cast's agents at build time
+(the same seam as `agent.manifest`), which arms verdict validation in the base agent.
+The machine-readable verdict and run-summary keys this produces are documented in
+[events.md](events.md#verdict-and-run-payloads-adr-0029).
+
 ## Scheduling lives on the agents
 
 A scenario does **not** declare a scheduling policy.  Cadence is per-agent —

modal/README.md

@@ -71,6 +71,11 @@ sizing, and how to add models/providers or wire endpoints into the engine.
   radius; one provider's outage or redeploy never touches another.
 - **Scalable** — serverless autoscaling, input concurrency, a shared weight
   cache (pull once, warm everywhere), and per-model `min_containers` warm pools.
+- **Fast cold starts** — snapshot-enabled models (`gpu_snapshot=True`) restore a
+  pre-warmed engine from a Modal memory snapshot in seconds instead of re-paying
+  download + load + warmup; `MODAL_LLM_KEEP_WARM=1` at deploy time pins warm
+  containers for the tier models on demo day. See
+  [`docs/deploying.md` → Cold starts](docs/deploying.md#cold-starts) (ADR-0030).
 - **Extensible** — add a model = one `ModelConfig` in `catalogue.py`; add a
   provider = one `Provider` entry + one app file. The serving path is written once
   in `service.py`, and the engine picks up the new model with no edits (it reads

modal/catalogue.py

@@ -82,6 +82,18 @@ class ModelConfig:
     max_num_seqs: int | None = None  # cap sequences batched per step (memory vs. throughput)
     max_num_batched_tokens: int | None = None  # token budget per scheduler step (prefill throughput)
 
+    # Cold starts. Opt a model into Modal memory snapshots (CPU + experimental GPU
+    # snapshot): the container boots once, loads weights, warms the engine, puts it
+    # to sleep (vLLM sleep mode, weights offloaded to host RAM), and is snapshotted;
+    # every later cold start restores the snapshot and wakes the engine in seconds
+    # instead of re-paying download + load + warmup. Constraints (why this is per
+    # model, not global): single-GPU models only, the model's vLLM build must
+    # support `--enable-sleep-mode`, and host RAM must hold the offloaded weights.
+    # Modal marks GPU snapshots alpha — keep it off for exotic serving paths
+    # (Transformers-backend Gemma, the omni specialist) and flip off on any model
+    # that misbehaves; the plain serving path is unchanged.
+    gpu_snapshot: bool = False
+
     # Observability / request logging (vLLM serve flags). Defaults give per-request
     # visibility in the container logs out of the box; see ``service.build_command``.
     log_requests: bool = True  # log each request's id, sampling params, and token counts
@@ -153,12 +165,17 @@ NVIDIA_MODELS: tuple[ModelConfig, ...] = (
         trust_remote_code=True,
         gated=True,
         max_concurrent_inputs=32,
+        # Tiny tier is the cast's hottest endpoint and 4B of BF16 weights (~8GB)
+        # easily fit host RAM during sleep — the ideal snapshot candidate.
+        gpu_snapshot=True,
     ),
     ModelConfig(
         name="nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16",
         endpoint_name="nemotron-3-nano-30b",
         # 30B total params in BF16 (~60GB) though only ~3B activate per token.
         # An alternate strong model — not cast to a profile by default.
+        # No gpu_snapshot: sleep mode would offload ~60GB of weights to host RAM,
+        # past what a default container comfortably holds.
         params_b=30,
         gpu="H200:1",
         max_model_len=32768,
@@ -187,6 +204,9 @@ NVIDIA_MODELS: tuple[ModelConfig, ...] = (
         tool_call_parser="hermes",
         enable_auto_tool_choice=True,
         max_concurrent_inputs=48,
+        # Qwen3-native single-GPU path on the pinned vLLM — snapshot-safe, and a
+        # reasoning model is exactly where a multi-minute cold start hurts most.
+        gpu_snapshot=True,
     ),
 )
 
@@ -202,6 +222,10 @@ OPENBMB_MODELS: tuple[ModelConfig, ...] = (
         max_model_len=32768,
         trust_remote_code=True,
         max_concurrent_inputs=48,
+        # Fast tier default for the cast; 8B BF16 (~16GB) offloads to host RAM
+        # fine. Sleep mode is allocator-level, so the custom MiniCPM modeling
+        # code doesn't affect it.
+        gpu_snapshot=True,
         # No tool_call_parser on purpose: MiniCPM4.1 emits a custom
         # <|tool_call_start|> format vLLM 0.21.0 has no parser for, so tool-call
         # structured output 400s here. The engine's structured path uses vLLM
@@ -252,7 +276,10 @@ GOOGLE_MODELS: tuple[ModelConfig, ...] = (
         # Served via vLLM's Transformers modeling backend (gemma4_unified has no
         # native vLLM class), which runs eager-only — CUDA-graph capture and the
         # async scheduler aren't supported on that path, so disable both here.
-        # Prefix caching still applies and stays on (the default).
+        # Prefix caching still applies and stays on (the default). gpu_snapshot
+        # stays off too: sleep mode on the nightly Transformers backend is
+        # unverified, and the Gemmas already skip the costliest warmup (no
+        # CUDA-graph capture).
         enforce_eager=True,
         async_scheduling=False,
         # Text-only in the cast (vision/audio is the MiniCPM-o specialist's job).

modal/docs/deploying.md

@@ -79,6 +79,7 @@ changes needed:
 | `target_concurrent_inputs` | Autoscale target — scale out here, burst to the max (defaults to ~75% of the ceiling). |
 | `buffer_containers`     | Extra idle containers pre-warmed under active load (bursty traffic). |
 | `scaledown_window`      | Idle seconds before a container stops (cold-start vs. cost).   |
+| `gpu_snapshot`          | Serve via Modal memory snapshots (CPU + GPU): cold starts restore a warmed engine in seconds instead of re-paying load + warmup. See [Cold starts](#cold-starts). |
 | `min_containers`        | Keep N warm to eliminate cold starts (always-on cost).         |
 | `gpu_memory_utilization` | Fraction of VRAM for weights + KV cache (vLLM default `0.9`); raise for a bigger KV cache. |
 | `enable_prefix_caching` | Reuse the KV cache for shared prompt prefixes (on by default — big win when the system prompt / ledger context repeats across the cast). |
@@ -131,6 +132,46 @@ per model:
 For memory-bound models, raise `gpu_memory_utilization` (more KV cache → more
 concurrency) and cap `max_num_seqs` / `max_num_batched_tokens` if a step OOMs.
 
+### Cold starts
+
+A scale-from-zero cold start normally pays the full pipeline: container boot →
+weight load → engine warmup — minutes for the bigger models. Two mechanisms cut
+this (ADR-0030):
+
+**1. Memory snapshots (`gpu_snapshot=True`, per model).** The first container
+boots once, loads weights, runs a few warmup completions, puts vLLM to sleep
+(sleep level 1: weights offloaded to host RAM, KV cache dropped), and Modal
+snapshots the container — CPU *and* GPU state. Every later cold start restores
+the snapshot and wakes the engine, turning a multi-minute boot into seconds.
+Under the hood this switches the model from the plain `@app.function` web server
+to a class-based lifecycle (`@modal.enter(snap=True)` warmup → snapshot →
+`@modal.enter(snap=False)` wake), but the public URL and API are identical —
+clients can't tell the paths apart.
+
+Snapshot-enabled today: `nemotron-3-nano-4b` (tiny), `minicpm-4-1-8b` (fast),
+`nemotron-cascade-14b`. Left off deliberately: the Gemmas (nightly
+Transformers-backend path, sleep mode unverified), `nemotron-3-nano-30b`
+(~60GB of weights won't fit host RAM during sleep), and the omni specialist.
+GPU snapshots are **Modal-alpha** — if a snapshot model misbehaves, set its
+`gpu_snapshot=False` and redeploy; the plain path is unchanged.
+
+**2. Demo-day keep-warm (deploy-time, no code edits).** Pin warm containers for
+every *profile-bound* model (tiny/fast/balanced/strong) right before a live
+demo — specialists keep scale-to-zero:
+
+```bash
+MODAL_LLM_KEEP_WARM=1 modal deploy modal/app_nvidia.py   # one warm container per tier model
+modal deploy modal/app_nvidia.py                         # back to scale-to-zero after
+```
+
+This burns GPU-hours while deployed; it's a switch for the hours around a demo,
+not a steady state. `min_containers` in `catalogue.py` remains the per-model
+override for anything finer-grained.
+
+Cold-start clients must follow redirects: a Modal endpoint that hasn't answered
+within ~150s returns a `303` to the same URL while the container finishes
+booting (`modal/healthcheck.py` handles this; so does the engine's gateway).
+
 ### Add a model
 
 Append one `ModelConfig` to the appropriate provider list in `catalogue.py` (tag

modal/service.py

@@ -80,6 +80,13 @@ JSON_LOGS = os.environ.get("MODAL_LLM_JSON_LOGS", "").lower() in ("1", "true", "
 # config applies the same level). Read at deploy time and baked into the image.
 LOG_LEVEL = os.environ.get("MODAL_LLM_LOG_LEVEL", "INFO").upper()
 
+# Demo-day switch: keep N containers warm for every *profile-bound* model (the
+# tiers the cast actually runs on), removing their cold starts entirely for the
+# duration of the deploy. Specialists keep scale-to-zero. Costs GPU-hours while
+# deployed — turn it on right before a live demo, redeploy without it after:
+#   MODAL_LLM_KEEP_WARM=1 modal deploy modal/app_nvidia.py
+KEEP_WARM = int(os.environ.get("MODAL_LLM_KEEP_WARM", "0") or "0")
+
 # Where the structured-logging module + its generated config live in the
 # container. The module dir goes on PYTHONPATH so vLLM can import the formatter
 # the dictConfig references (``vllm_logging.JsonFormatter``).
@@ -139,6 +146,12 @@ def build_image(cfg: ModelConfig) -> modal.Image:
             .env({"PYTHONPATH": _LOG_MODULE_DIR})
             .env({"MODAL_LLM_JSON_LOGS": "1", "MODAL_LLM_LOG_LEVEL": LOG_LEVEL})
         )
+    if cfg.gpu_snapshot:
+        # Snapshot prerequisites: VLLM_SERVER_DEV_MODE exposes the /sleep and
+        # /wake_up endpoints the snapshot lifecycle drives, and single-threaded
+        # inductor compilation keeps torch.compile artifacts snapshot-safe
+        # (Modal's documented vLLM + GPU-snapshot recipe).
+        image = image.env({"VLLM_SERVER_DEV_MODE": "1", "TORCHINDUCTOR_COMPILE_THREADS": "1"})
     if cfg.extra_pip:
         image = image.uv_pip_install(*cfg.extra_pip)
     if cfg.env:
@@ -204,6 +217,10 @@ def build_command(cfg: ModelConfig) -> list[str]:
         cmd += ["--tool-call-parser", cfg.tool_call_parser]
     if cfg.mm_limits:
         cmd += ["--limit-mm-per-prompt", json.dumps(cfg.mm_limits)]
+    if cfg.gpu_snapshot:
+        # Sleep mode lets the snapshot lifecycle offload weights to host RAM
+        # (sleep level 1) before the memory snapshot is taken, then wake on restore.
+        cmd += ["--enable-sleep-mode"]
     cmd += list(cfg.extra_vllm_args)
     return cmd
 
@@ -211,12 +228,19 @@ def build_command(cfg: ModelConfig) -> list[str]:
 # --- Endpoint registration ------------------------------------------------------
 
 
-def register_model(app: modal.App, cfg: ModelConfig) -> modal.Function:
+def register_model(app: modal.App, cfg: ModelConfig) -> modal.Function | type:
     """Attach one model to ``app`` as an autoscaling, OpenAI-compatible endpoint.
 
-    The function is serialized (its prebuilt ``vllm serve`` argv is shipped to
-    the container), which lets us register many distinctly-named endpoints from
-    a simple loop without each needing a hand-written module-level function.
+    Dispatches on ``cfg.gpu_snapshot``: the default path is a serialized
+    ``@app.function`` web server; snapshot models use a class-based lifecycle
+    (load → warm up → sleep → snapshot) so later cold starts restore in seconds
+    instead of re-paying download + load + warmup. Both paths publish the same
+    URL shape (``…--<app>-<endpoint_name>.modal.run``), so clients can't tell
+    them apart.
+
+    Everything is serialized (the prebuilt ``vllm serve`` argv is shipped to the
+    container), which lets us register many distinctly-named endpoints from a
+    simple loop without each needing a hand-written module-level function.
     """
     image = build_image(cfg)
     cmd = build_command(cfg)
@@ -227,11 +251,28 @@ def register_model(app: modal.App, cfg: ModelConfig) -> modal.Function:
         # Exposes VLLM_API_KEY in the container; vLLM then enforces bearer auth.
         secrets.append(modal.Secret.from_name(API_KEY_SECRET_NAME))
 
+    # Demo-day keep-warm: pin warm containers for the tier-bound models only —
+    # specialists keep scale-to-zero (see KEEP_WARM above).
+    min_containers = cfg.min_containers
+    if KEEP_WARM and cfg.profile:
+        min_containers = max(min_containers, KEEP_WARM)
+
     # Autoscale at the target, but let a hot container absorb a burst up to the
     # hard max before another cold-starts (Modal high-perf-inference guidance).
     # Default the target to ~75% of the ceiling so we scale out before saturating.
     target_inputs = cfg.target_concurrent_inputs or max(1, (cfg.max_concurrent_inputs * 3) // 4)
 
+    if cfg.gpu_snapshot:
+        return _register_snapshot_model(
+            app,
+            cfg,
+            image=image,
+            cmd=cmd,
+            secrets=secrets,
+            min_containers=min_containers,
+            target_inputs=target_inputs,
+        )
+
     function_kwargs = dict(
         name=cfg.endpoint_name,
         image=image,
@@ -239,7 +280,7 @@ def register_model(app: modal.App, cfg: ModelConfig) -> modal.Function:
         volumes={HF_CACHE_PATH: hf_cache_vol, VLLM_CACHE_PATH: vllm_cache_vol},
         secrets=secrets,
         scaledown_window=cfg.scaledown_window,
-        min_containers=cfg.min_containers,
+        min_containers=min_containers,
         timeout=cfg.request_timeout,
         serialized=True,
     )
@@ -270,6 +311,138 @@ def register_model(app: modal.App, cfg: ModelConfig) -> modal.Function:
     return serve
 
 
+def _class_name(slug: str) -> str:
+    """Modal class name for an endpoint slug: ``nemotron-3-nano-4b`` → ``Nemotron3Nano4b``."""
+    return "".join(part.capitalize() for part in slug.replace("_", "-").split("-") if part) or "SnapshotServer"
+
+
+def _register_snapshot_model(
+    app: modal.App,
+    cfg: ModelConfig,
+    *,
+    image: modal.Image,
+    cmd: list[str],
+    secrets: list[modal.Secret],
+    min_containers: int,
+    target_inputs: int,
+) -> type:
+    """Snapshot serving path — Modal's vLLM + GPU-memory-snapshot recipe.
+
+    First boot: start vLLM, wait for the port, run a few warmup completions so
+    compiled artifacts and caches are resident, put the engine to sleep (weights
+    offloaded to host RAM, KV cache dropped), and let Modal snapshot the
+    container (CPU + GPU state). Every later cold start restores the snapshot
+    and wakes the engine — seconds instead of minutes. The web URL label is
+    pinned to ``cfg.endpoint_name`` so the public URL is identical to the plain
+    function path (``…--<app>-<endpoint_name>.modal.run``).
+    """
+    served_name = cfg.served_name
+
+    # Helpers are nested (not module-level) on purpose: the class ships to the
+    # container via cloudpickle (``serialized=True``), and closures are pickled
+    # by value — a module-level helper would be pickled by reference to the
+    # ``service`` module, which doesn't exist inside the container.
+    def _headers() -> dict[str, str]:
+        import os
+
+        key = os.environ.get("VLLM_API_KEY")
+        return {"Authorization": f"Bearer {key}"} if key else {}
+
+    def _wait_ready(proc) -> None:
+        # vLLM opens the port only once the engine is initialized, so a
+        # successful connect means "ready", not just "listening".
+        import socket
+        import time
+
+        while True:
+            try:
+                socket.create_connection(("localhost", VLLM_PORT), timeout=1).close()
+                return
+            except OSError:
+                if proc.poll() is not None:
+                    raise RuntimeError(f"vllm exited with code {proc.returncode}")
+                time.sleep(0.2)
+
+    def _post(path: str, json_body: dict | None = None, timeout: float = 300.0) -> None:
+        import requests  # vLLM dependency, always present in the image
+
+        url = f"http://localhost:{VLLM_PORT}{path}"
+        requests.post(url, headers=_headers(), json=json_body, timeout=timeout).raise_for_status()
+
+    class _SnapshotServer:
+        @modal.enter(snap=True)
+        def start(self):
+            import os
+            import subprocess
+
+            env = dict(os.environ)
+            # Same structured-logging hook as the plain path (see ``serve``).
+            if env.get("MODAL_LLM_JSON_LOGS", "").lower() in ("1", "true", "yes"):
+                import vllm_logging
+
+                vllm_logging.write_config(_LOG_CONFIG_PATH, level=env.get("MODAL_LLM_LOG_LEVEL", "INFO"))
+                env["VLLM_LOGGING_CONFIG_PATH"] = _LOG_CONFIG_PATH
+
+            self.vllm_proc = subprocess.Popen(cmd, env=env)
+            _wait_ready(self.vllm_proc)
+            # Touch the full serving path so compile/caching work happens *before*
+            # the snapshot rather than on the first real request after restore.
+            warmup = {
+                "model": served_name,
+                "messages": [{"role": "user", "content": "Who tends the wood?"}],
+                "max_tokens": 8,
+            }
+            for _ in range(3):
+                _post("/v1/chat/completions", json_body=warmup)
+            # Offload weights to host RAM (sleep level 1); Modal snapshots the
+            # container right after the snap=True enters return.
+            _post("/sleep?level=1", timeout=120.0)
+
+        @modal.enter(snap=False)
+        def wake(self):
+            # Runs after every restore (and on the snapshot-creating boot itself,
+            # which simply resumes serving): reload weights onto the GPU.
+            _post("/wake_up", timeout=120.0)
+            _wait_ready(self.vllm_proc)
+
+        @modal.web_server(port=VLLM_PORT, startup_timeout=cfg.startup_timeout, label=cfg.endpoint_name)
+        def serve(self):
+            pass  # vLLM (already running) is the web server; Modal just exposes the port.
+
+        @modal.exit()
+        def stop(self):
+            proc = getattr(self, "vllm_proc", None)
+            if proc is not None:
+                proc.terminate()
+
+    # One Modal class per model, named after the endpoint (App.cls has no name
+    # override, so rename the type before decorating).
+    name = _class_name(cfg.endpoint_name)
+    _SnapshotServer.__name__ = name
+    _SnapshotServer.__qualname__ = name
+
+    cls_kwargs = dict(
+        image=image,
+        gpu=cfg.gpu,
+        volumes={HF_CACHE_PATH: hf_cache_vol, VLLM_CACHE_PATH: vllm_cache_vol},
+        secrets=secrets,
+        scaledown_window=cfg.scaledown_window,
+        min_containers=min_containers,
+        timeout=cfg.request_timeout,
+        # Bounds the whole snap=True phase (download + load + warmup + sleep).
+        startup_timeout=cfg.startup_timeout,
+        serialized=True,
+        enable_memory_snapshot=True,
+        # GPU snapshots are Modal-alpha; scoped per model via cfg.gpu_snapshot.
+        experimental_options={"enable_gpu_snapshot": True},
+    )
+    if cfg.buffer_containers:
+        cls_kwargs["buffer_containers"] = cfg.buffer_containers
+
+    concurrent = modal.concurrent(max_inputs=cfg.max_concurrent_inputs, target_inputs=target_inputs)
+    return app.cls(**cls_kwargs)(concurrent(_SnapshotServer))
+
+
 def register_all(app: modal.App, configs: Iterable[ModelConfig]) -> None:
     """Register every model in ``configs`` onto ``app``."""
     for cfg in configs:

src/agents/base.py

@@ -115,6 +115,12 @@ class ManifestAgent(Agent):
         self.memory_index = memory_index
         self._reflection_tracker: ReflectionTracker | None = None
         self.last_usage: dict[str, int] = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
+        # Scenario competition context (ADR-0029), injected by the registry when this
+        # agent is assembled into a cast.  ``None`` (the standalone default) means no
+        # competition: the verdict-validation hook is inert, so non-judged scenarios
+        # and bare-constructed agents behave exactly as before.
+        self.competition = None
+        self.cast_names: list[str] = []
         # The model behind the most recent generation, captured when the provider is
         # resolved and stamped onto the event in act()/reflection — so each line in the
         # ledger records the model that actually produced it, not just the intended one.
@@ -241,6 +247,10 @@ class ManifestAgent(Agent):
         Offline path (deterministic stub, no ``complete_structured``): append the
         JSON instruction and run the tolerant parser as before.  Token/cost usage
         is recorded from the provider in every path.
+
+        On either path, a judge's verdict in a competition scenario is run through
+        :meth:`_verify_verdict` — one corrective re-ask when the model names a winner
+        outside the cast (ADR-0029), otherwise a no-op for every other agent.
         """
         wants_thought = bool(extra_fields and "thought" in extra_fields)
         provider = self.router.for_profile(self._route_key)
@@ -248,25 +258,113 @@ class ManifestAgent(Agent):
         with obs.span("agent.resolve", **{"mal.agent": role, "mal.profile": self._route_key}):
             if hasattr(provider, "complete_structured"):
                 model = build_output_model(allowed, extra_fields)
-                try:
-                    result = provider.complete_structured(role, prompt, model)
+
+                def _structured(p: str) -> dict | None:
+                    """One structured generation: a usable payload, or None to fall back."""
+                    try:
+                        result = provider.complete_structured(role, p, model)
+                    except Exception:
+                        self.last_usage = dict(provider.last_usage)
+                        return None  # structured failed — caller falls through to prose
                     self.last_usage = dict(provider.last_usage)
                     payload = self._with_reasoning(result.model_dump(), provider, "", wants_thought)
-                    if is_usable_line(payload.get("text", "")):
-                        obs.add_span_attrs(**{"resolve.path": "structured", "event.kind": payload.get("kind", "")})
-                        return payload
-                except Exception:
-                    pass  # structured failed — fall through to the prose fallback
+                    return payload if is_usable_line(payload.get("text", "")) else None
+
+                payload = _structured(prompt)
+                if payload is not None:
+                    payload = self._verify_verdict(prompt, payload, _structured)
+                    obs.add_span_attrs(**{"resolve.path": "structured", "event.kind": payload.get("kind", "")})
+                    return payload
                 obs.add_span_attrs(**{"resolve.path": "prose_fallback"})
                 return self._prose_fallback(role, prompt, allowed, wants_thought, provider)
 
             instruction = json_instruction(allowed, extra_fields=extra_fields)
-            raw = provider.complete(role, f"{prompt}\n{instruction}")
-            self.last_usage = dict(provider.last_usage)
-            self._guard_model_error(role, raw)
-            parsed = parse_agent_output(raw, allowed_kinds=allowed, fallback_kind=allowed[0])
+
+            def _offline(p: str) -> dict:
+                """One offline generation: parse the stub's output into a payload."""
+                raw = provider.complete(role, f"{p}\n{instruction}")
+                self.last_usage = dict(provider.last_usage)
+                self._guard_model_error(role, raw)
+                parsed = parse_agent_output(raw, allowed_kinds=allowed, fallback_kind=allowed[0])
+                return self._with_reasoning(parsed, provider, raw, wants_thought)
+
+            parsed = self._verify_verdict(prompt, _offline(prompt), _offline)
             obs.add_span_attrs(**{"resolve.path": "offline_parse", "event.kind": parsed.get("kind", "")})
-            return self._with_reasoning(parsed, provider, raw, wants_thought)
+            return parsed
+
+    # ── verdict winner validation (ADR-0029) ───────────────────────────────────
+
+    def _verify_verdict(self, prompt: str, payload: dict, regenerate) -> dict:
+        """Validate a verdict's ``winner``/``scores``; re-ask once on a bad winner.
+
+        ``scores`` is normalised in place (non-cast keys dropped, values clamped to
+        0–10) and never re-asked — it is garnish.  An out-of-cast ``winner`` triggers
+        exactly one corrective regeneration via *regenerate*, with the token usage of
+        both calls summed so the governor (ADR-0013) meters the retry.  A second
+        failure drops ``winner`` and stamps ``no_contest`` — the verdict *text* still
+        ships, so the show always ends; only the leaderboard row is forfeited.
+
+        For every non-judge agent, every ``kind: none`` scenario, and every standalone
+        agent (no competition attached), :meth:`_validate_payload` returns ``None`` and
+        this is a transparent pass-through."""
+        error = self._validate_payload(payload)
+        if error is None:
+            return payload
+        first_usage = dict(self.last_usage)
+        corrective = (
+            f"\n\nCORRECTION: your previous reply named an invalid winner. {error} "
+            "Reply again with the same JSON object, changing only the 'winner' field."
+        )
+        retry = regenerate(prompt + corrective)
+        self.last_usage = self._sum_usage(first_usage, self.last_usage)
+        if retry is not None and is_usable_line(retry.get("text", "")) and self._validate_payload(retry) is None:
+            return retry
+        payload.pop("winner", None)
+        payload["no_contest"] = True
+        return payload
+
+    def _validate_payload(self, parsed: dict) -> str | None:
+        """Return an error string when a judge named an out-of-cast winner, else ``None``.
+
+        Active only for a ``judge`` whose attached competition has ``kind != none`` and
+        whose manifest declares a ``winner`` field.  A missing/empty ``winner`` is *not*
+        an error (the field is optional and the offline stub never emits it — determinism
+        preserved); ``scores`` is normalised in place as a side effect (never an error)."""
+        comp = self.competition
+        if comp is None or getattr(comp, "kind", "none") == "none" or self.manifest.role != "judge":
+            return None
+        if "winner" not in (self.manifest.output_extra_fields or []):
+            return None
+        cast = set(self.cast_names)
+        scores = parsed.get("scores")
+        if isinstance(scores, dict):
+            cleaned: dict[str, float] = {}
+            for name, value in scores.items():
+                if name not in cast:
+                    continue
+                try:
+                    cleaned[name] = max(0.0, min(10.0, float(value)))
+                except (TypeError, ValueError):
+                    continue
+            parsed["scores"] = cleaned
+        winner = parsed.get("winner")
+        if winner in (None, ""):
+            return None
+        if winner not in self._winner_vocab():
+            return f"'winner' must be exactly one of: {', '.join(self._winner_vocab())}."
+        return None
+
+    def _winner_vocab(self) -> list[str]:
+        """The valid winner vocabulary: every cast member, plus any team labels."""
+        comp = self.competition
+        teams = getattr(comp, "teams", None) or {}
+        return list(self.cast_names) + list(teams.keys())
+
+    @staticmethod
+    def _sum_usage(first: dict, second: dict) -> dict:
+        """Sum two token-usage dicts so a re-ask's cost is metered, not lost."""
+        keys = set(first) | set(second)
+        return {k: int(first.get(k, 0) or 0) + int(second.get(k, 0) or 0) for k in keys}
 
     def _guard_model_error(self, role: str, raw: str) -> None:
         """Raise when *raw* is a provider failure sentinel, not a spoken line.

src/agents/handlers.py

@@ -55,8 +55,53 @@ class SpyHost(ManifestAgent):
         ]
         if reveal:
             event.payload["reveal"] = reveal
+        self._stamp_scoreboard(event)
         return event
 
+    def _stamp_scoreboard(self, event: Event) -> None:
+        """Score the verdict in code — the load-bearing split of ADR-0029.
+
+        The judge's prose names a suspect (``payload['winner']`` on the live path);
+        this handler turns that *accusation* into the ground-truth *result* using the
+        scenario's ``competition.teams``: the herd wins when the named player really is
+        a spy, the spy wins otherwise.  The accusation is preserved as ``accused`` and a
+        ``correct`` flag rides alongside, so the trace stays auditable.  Offline (no
+        ``winner`` field) the accusation is recovered from the verdict text, so the
+        no-API-key demo still ends on a full, deterministic scoreboard.  With no spy
+        team declared, or no recoverable accusation, the round is a ``no_contest``.
+        """
+        comp = self.competition
+        spies = set((getattr(comp, "teams", None) or {}).get("spy", []))
+        if getattr(comp, "kind", "none") != "versus" or not spies:
+            return
+        accused = event.payload.get("winner") or self._scan_accusation(str(event.payload.get("text", "")))
+        if not accused:
+            event.payload.pop("winner", None)
+            event.payload["no_contest"] = True
+            return
+        correct = accused in spies
+        event.payload["accused"] = accused
+        event.payload["correct"] = correct
+        event.payload["winner"] = "herd" if correct else "spy"
+
+    def _scan_accusation(self, text: str) -> str | None:
+        """Recover the accused player from verdict *text* — the first cast name named.
+
+        Matches each player by the distinctive tail of its agent name (``spy-cara`` →
+        ``cara``), case-insensitively, and returns the one mentioned earliest.  The host
+        itself is excluded so it never accuses the judge."""
+        low = text.lower()
+        best: str | None = None
+        best_at = len(low) + 1
+        for name in self.cast_names:
+            if name == self.name:
+                continue
+            token = name.split("-")[-1].lower()
+            at = low.find(token) if token else -1
+            if at != -1 and at < best_at:
+                best, best_at = name, at
+        return best
+
 
 @register_handler("fortune-teller")
 class FortuneTeller(ManifestAgent):

src/core/conductor.py

@@ -155,12 +155,20 @@ class Conductor:
         *,
         winner: str | None = None,
         winning_model: str | None = None,
+        winner_kind: str | None = None,
+        winning_models: list[str] | None = None,
     ) -> Event | None:
         """Close the current run with a ``run.finished`` event.
 
         Idempotent-safe: if this run already has a ``run.finished`` event we return
         the existing one rather than emitting a duplicate.  ``turns`` and ``tokens``
         are read from the governor's live counters.
+
+        Attribution (ADR-0029): ``winner`` is a cast agent name (``winner_kind:
+        "agent"``) or a team label (``winner_kind: "team"``).  ``winning_model`` keeps
+        its original meaning — a single cast agent's endpoint, populated only for an
+        agent winner — while ``winning_models`` lists the endpoint(s) behind the
+        winner (every member of a winning team).  All keys are additive.
         """
         existing = [e for e in self.ledger.events_for_run(self.run_id) if e.kind == "run.finished"]
         if existing:
@@ -174,7 +182,9 @@ class Conductor:
             payload={
                 "reason": reason,
                 "winner": winner,
+                "winner_kind": winner_kind,
                 "winning_model": winning_model,
+                "winning_models": list(winning_models or []),
                 "turns": int(stats.get("current_turn", self.turn) or self.turn),
                 "tokens": int(stats.get("total_tokens", 0) or 0),
             },
@@ -184,6 +194,7 @@ class Conductor:
             run_id=self.run_id,
             reason=reason,
             winner=winner,
+            winner_kind=winner_kind,
             winning_model=winning_model,
             turns=finished.payload["turns"],
             tokens=finished.payload["tokens"],

src/core/config.py

@@ -16,8 +16,11 @@ to "emit JSON, validate it, run it."  See ADR-0011.
 The agent schema itself is :class:`AgentManifest` (``src/core/manifest.py``) — we
 reuse it here rather than duplicating, so the four stable contracts stay singular.
 """
+
 from __future__ import annotations
 
+from typing import Literal
+
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from src.core.manifest import AgentManifest
@@ -75,6 +78,56 @@ class GovernorConfig(BaseModel):
     hourly_budget_usd: float | None = None
 
 
+# ── competition ──────────────────────────────────────────────────────────────────
+
+
+class CompetitionConfig(BaseModel):
+    """Declares whether — and how — a scenario produces a winner (ADR-0029).
+
+    A scenario can be a ``versus`` contest between named teams, a ``judged`` pick
+    where the judge's verdict *is* the result, or ``none`` (the default) where
+    nobody wins.  ``winner`` downstream carries either an agent name or a team
+    label, so the team labels here must stay distinct from agent names — that
+    cross-cast check lives in :meth:`WorldConfig._check_cast_references`, while the
+    rules a competition can enforce on its own (team shape, disjointness) live in
+    the validator below.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    kind: Literal["versus", "judged", "none"] = "none"
+    """How a winner is derived — ``versus`` (team contest), ``judged`` (the judge's
+    pick is the answer), or ``none`` (no winner; the default and the absent block)."""
+
+    teams: dict[str, list[str]] | None = None
+    """Team label → member agent names.  Permitted only when ``kind == 'versus'``."""
+
+    @model_validator(mode="after")
+    def _check_teams(self) -> "CompetitionConfig":
+        if self.kind != "versus":
+            if self.teams is not None:
+                raise ValueError(f"competition.teams is only allowed when kind is 'versus' (got kind={self.kind!r})")
+            return self
+        # kind == "versus": teams are required and must describe a real contest.
+        if not self.teams:
+            raise ValueError("competition.kind 'versus' requires a non-empty 'teams' mapping")
+        empty = [label for label, members in self.teams.items() if not members]
+        if empty:
+            raise ValueError(f"competition.teams has empty member lists for teams: {sorted(empty)}")
+        seen: dict[str, str] = {}
+        overlap: set[str] = set()
+        for label, members in self.teams.items():
+            for member in members:
+                if member in seen and seen[member] != label:
+                    overlap.add(member)
+                seen[member] = label
+        if overlap:
+            raise ValueError(
+                f"competition.teams must be mutually disjoint; agents on more than one team: {sorted(overlap)}"
+            )
+        return self
+
+
 # ── scenario ─────────────────────────────────────────────────────────────────────
 
 
@@ -96,6 +149,9 @@ class ScenarioConfig(BaseModel):
     genesis_text: str | None = None
     governor: GovernorConfig | None = None
 
+    competition: CompetitionConfig | None = None
+    """Optional winner contract (ADR-0029); absent == ``none`` (no winner)."""
+
 
 # ── the whole world ──────────────────────────────────────────────────────────────
 
@@ -125,6 +181,25 @@ class WorldConfig(BaseModel):
                     f"scenario {scenario.name!r} references undefined agents: {missing}. "
                     f"Defined agents: {sorted(defined)}"
                 )
+            competition = scenario.competition
+            if competition is None or competition.teams is None:
+                continue
+            # Every team member must be in this scenario's cast (ADR-0029 §1).
+            cast = set(scenario.cast)
+            off_cast = sorted({m for members in competition.teams.values() for m in members if m not in cast})
+            if off_cast:
+                raise ValueError(
+                    f"scenario {scenario.name!r} competition team members not in its cast: {off_cast}. "
+                    f"Cast: {sorted(cast)}"
+                )
+            # A team label must not collide with any agent name, or the winner union
+            # (agent name OR team label) becomes ambiguous (ADR-0029 §1).
+            collisions = sorted(label for label in competition.teams if label in defined)
+            if collisions:
+                raise ValueError(
+                    f"scenario {scenario.name!r} competition team labels collide with agent names: {collisions}. "
+                    f"Team labels must be distinct from agent names to keep the winner unambiguous."
+                )
         return self
 
 

src/core/registry.py

@@ -239,6 +239,13 @@ class Registry:
 
         memory_index = memory_index_from_env()
         agents = tuple(self.build_agent(agent_name, router, tools, memory_index) for agent_name in cfg.cast)
+        # Inject the scenario's competition context (ADR-0029) — the same single-attribute
+        # seam as ``agent.manifest``.  This is the only scenario-level fact an agent sees:
+        # it lets a judge validate its winner against the cast and a versus handler attribute
+        # the win to a team.  Absent block == no competition, so the hook stays inert.
+        for agent in agents:
+            agent.competition = cfg.competition
+            agent.cast_names = list(cfg.cast)
         obs.log(
             "registry.cast_assembled",
             scenario=cfg.name,

src/core/run_index.py

@@ -56,7 +56,14 @@ class RunSummary(BaseModel):
     finished_at: datetime | None = None
     reason: str | None = None
     winner: str | None = None
+    winner_kind: str | None = None
+    """Whether ``winner`` names a cast agent (``"agent"``) or a team label
+    (``"team"``); ``None`` for runs with no competition or no winner (ADR-0029)."""
     winning_model: str | None = None
+    winning_models: list[str] = Field(default_factory=list)
+    """Endpoint(s) behind the winner — the agent's model, or every member of a
+    winning team (``None`` entries dropped).  ``winning_model`` stays the single
+    agent-winner endpoint for back-compat."""
     turns: int = 0
     tokens: int = 0
 
@@ -88,7 +95,9 @@ def _apply_finished(summary: RunSummary, event: Event) -> None:
     payload = event.payload
     summary.reason = payload.get("reason")
     summary.winner = payload.get("winner")
+    summary.winner_kind = payload.get("winner_kind")
     summary.winning_model = payload.get("winning_model")
+    summary.winning_models = list(payload.get("winning_models") or [])
     summary.turns = int(payload.get("turns", 0) or 0)
     summary.tokens = int(payload.get("tokens", 0) or 0)
     summary.finished_at = event.created_at

src/core/structured.py

@@ -40,6 +40,34 @@ class AgentOutputError(ValueError):
     """Raised when output cannot be normalised to a valid event payload."""
 
 
+# ── well-known typed extra fields ──────────────────────────────────────────────
+
+
+# A small, curated table of *engine-known* extra fields (ADR-0029).  Most
+# ``output_extra_fields`` are arbitrary scenario fields (``mood``, ``wants``, …) and
+# remain required strings, but ``winner`` and ``scores`` are the verdict contract
+# ADR-0026 already names in ``run.finished`` — so the engine gives them real types:
+# an *optional* cast name and an *optional* ``{player: score}`` map.  Each entry pairs
+# the Pydantic field spec (built lazily so ``Field`` stays a local import) with the
+# JSON-schema hint ``json_instruction`` renders for that field.  Anything not in this
+# table hits the *other* row and behaves exactly as before — back-compat is total.
+def _well_known_specs() -> dict[str, tuple[Any, str]]:
+    """Return ``{field: (pydantic_spec, json_hint)}`` for engine-known extra fields.
+
+    Built behind a function so ``Field`` is a local import (matching the lazy-import
+    idiom of ``build_output_model``) and never touched on the offline path that
+    doesn't construct a validated model.
+    """
+    from pydantic import Field
+
+    # The hint is the literal JSON value to render after ``"<field>": `` — a quoted
+    # string for ``winner``, a bare object for ``scores`` (it is not a string value).
+    return {
+        "winner": ((str | None, None), '"<a player\'s name, or null>"'),
+        "scores": ((dict[str, float], Field(default_factory=dict)), '{"<player>": 0-10}'),
+    }
+
+
 # ── validated output model (live path) ─────────────────────────────────────────
 
 
@@ -50,22 +78,27 @@ def build_output_model(
     """Build a Pydantic model for an agent's validated output.
 
     ``kind`` is constrained to *allowed_kinds* via a ``Literal``, so the model
-    cannot emit a kind it is not authorised for; ``text`` plus any *extra_fields*
-    are required strings.  Used on the live path with structured output: the
-    provider retries on validation failure and returns a valid instance, which
-    means the malformed-prose ``_raw_fallback`` path is never taken.
+    cannot emit a kind it is not authorised for; ``text`` is a required string.
+    *extra_fields* are required strings too, *except* the **well-known typed
+    fields** of ADR-0029: ``winner`` becomes ``str | None`` (default ``None``) and
+    ``scores`` becomes ``dict[str, float]`` (default ``{}``).  Used on the live path
+    with structured output: the provider retries on validation failure and returns a
+    valid instance, which means the malformed-prose ``_raw_fallback`` path is never
+    taken.
 
     Args:
         allowed_kinds: event kinds this agent may emit (the ``may_emit`` grant,
             reflection excluded).  Must be non-empty.
-        extra_fields: optional additional payload fields (e.g. ``"emotion"``),
-            each a required string alongside ``text``.
+        extra_fields: optional additional payload fields (e.g. ``"emotion"``).  Each
+            is a required string alongside ``text`` unless it is a well-known typed
+            field (``winner``, ``scores``), which is optional with a typed default.
     """
     if not allowed_kinds:
         raise AgentOutputError("build_output_model requires at least one allowed kind")
 
     from pydantic import create_model
 
+    well_known = _well_known_specs()
     # A single-element Literal is legal and still constrains to that one kind.
     kind_type = Literal[tuple(allowed_kinds)]  # type: ignore[valid-type]
     fields: dict[str, Any] = {
@@ -73,7 +106,8 @@ def build_output_model(
         "text": (str, ...),
     }
     for name in extra_fields or []:
-        fields[name] = (str, ...)
+        spec, _hint = well_known.get(name, ((str, ...), None))
+        fields[name] = spec
 
     return create_model(
         "AgentOutput",
@@ -88,17 +122,40 @@ def build_output_model(
 def json_instruction(allowed_kinds: list[str], extra_fields: list[str] | None = None) -> str:
     """Return the JSON constraint block appended to every agent prompt.
 
+    For ordinary fields the schema hint is the uniform ``"...": "..."`` shape.  When
+    a **well-known typed field** of ADR-0029 is present, that field gets a typed hint
+    instead (``"winner": "<a player's name, or null>"``,
+    ``"scores": {"<player>": 0-10}``) so a small model knows it may answer ``null`` or
+    a number map rather than a sentence.  Manifests with no well-known field render
+    byte-identically to the original uniform schema.
+
     Args:
         allowed_kinds: event kinds this agent may emit.
         extra_fields: optional additional payload fields (e.g. "emotion", "wants").
     """
-    field_list = '", "'.join(["kind", "text"] + (extra_fields or []))
+    fields = extra_fields or []
+    well_known = _well_known_specs()
     kinds_str = " | ".join(allowed_kinds)
+
+    if any(name in well_known for name in fields):
+        # Richer per-field schema: typed hints for known fields, "..." for the rest.
+        # Only taken when a well-known field is present, so the common case below
+        # stays byte-identical to the original uniform-schema output.
+        hints = {"kind": '"..."', "text": '"..."'}
+        for name in fields:
+            _spec, hint = well_known.get(name, (None, None))
+            hints[name] = hint if hint is not None else '"..."'
+        schema_body = ", ".join(f'"{name}": {hints[name]}' for name in ["kind", "text", *fields])
+        schema_line = f"Schema: {{{schema_body}}}\n"
+    else:
+        field_list = '", "'.join(["kind", "text"] + list(fields))
+        schema_line = f'Schema: {{"{field_list}": "..."}}\n'
+
     return (
         "\n\nOUTPUT FORMAT\n"
         "Reply with a single JSON object and NOTHING else. No analysis, no reasoning, "
         "no <think> blocks, no markdown fences, no text before or after the JSON.\n"
-        f'Schema: {{"{field_list}": "..."}}\n'
+        f"{schema_line}"
         f"kind must be one of: {kinds_str}\n"
         "text must be one or two sentences, vivid and specific — your line, never your reasoning.\n"
         "If you were given a secret word, never spell or quote it; describe it only.\n"

src/models/provider.py

@@ -4,6 +4,7 @@ import hashlib
 import json
 import re
 from dataclasses import dataclass, field
+from typing import Any
 
 from src import observability as obs
 
@@ -256,7 +257,7 @@ class DeterministicTinyModel(ModelProvider):
             allowed_kinds, fields = schema
             extra = [f for f in fields if f not in ("kind", "text")]
             if extra:  # only agents that opted into extra fields take the JSON path
-                obj: dict[str, str] = {
+                obj: dict[str, Any] = {
                     "kind": allowed_kinds[int(digest[2:4], 16) % len(allowed_kinds)],
                     "text": text,
                 }
@@ -293,7 +294,7 @@ class DeterministicTinyModel(ModelProvider):
         obs.log("llm.exchange", level="debug", role=role, model=self.variant, prompt=prompt, completion=out)
         return out
 
-    def _synth_field(self, name: str, role: str, digest: str) -> str:
+    def _synth_field(self, name: str, role: str, digest: str) -> Any:
         """Deterministically synthesise a value for one requested extra field."""
         if name == "mood":
             moods = _STUB_MOODS_BY_ROLE.get(role, _STUB_MOODS)
@@ -301,5 +302,13 @@ class DeterministicTinyModel(ModelProvider):
         if name == "thought":
             opts = _STUB_THOUGHTS.get(role, _STUB_THOUGHT_DEFAULT)
             return opts[int(digest[6:8], 16) % len(opts)]
+        # Well-known verdict fields (ADR-0029) get their real types, not a placeholder:
+        # the stub names no winner (the field is optional, and a versus handler recovers
+        # the accusation from the verdict text), and emits an empty score map — so the
+        # offline path stays validation-clean and deterministic with no wasted re-ask.
+        if name == "winner":
+            return None
+        if name == "scores":
+            return {}
         # Unknown extra field: a short, stable placeholder keeps the output valid.
         return f"{name}:{digest[:4]}"

src/ui/fishbowl/session.py

@@ -115,11 +115,15 @@ class FishbowlSession:
     def finalize(self, reason: str) -> None:
         """Close the current run with a ``run.finished`` event (idempotent-safe).
 
-        On a verdict we derive ``winner`` from the judge's ruling (best-effort: the
-        ``winner`` payload key when present) and ``winning_model`` from the run.started
-        cast map; both fall back to ``None`` when unknown."""
+        On a verdict we derive ``winner`` from the judge's ruling and resolve its kind
+        (ADR-0029): a cast agent name → ``winner_kind: "agent"`` with that agent's
+        ``winning_model``; a team label → ``winner_kind: "team"`` with every member's
+        endpoint in ``winning_models`` (``winning_model`` left ``None``, never guessed).
+        All fall back to ``None`` / empty when unknown."""
         winner: str | None = None
+        winner_kind: str | None = None
         winning_model: str | None = None
+        winning_models: list[str] = []
         run_events = self.conductor.ledger.events_for_run(self.conductor.run_id)
         if reason == "verdict":
             verdict = next((e for e in reversed(run_events) if e.kind == "judge.verdict"), None)
@@ -128,8 +132,26 @@ class FishbowlSession:
             if winner:
                 started = next((e for e in run_events if e.kind == "run.started"), None)
                 cast = (started.payload.get("cast") or {}) if started is not None else {}
-                winning_model = (cast.get(winner) or {}).get("model_endpoint")
-        self.conductor.finalize(reason, winner=winner, winning_model=winning_model)
+                scenario = self._registry.scenarios.get(self._scenario_name)
+                teams = getattr(getattr(scenario, "competition", None), "teams", None) or {}
+                if winner in cast:
+                    winner_kind = "agent"
+                    winning_model = (cast.get(winner) or {}).get("model_endpoint")
+                    winning_models = [winning_model] if winning_model else []
+                elif winner in teams:
+                    winner_kind = "team"
+                    winning_models = [
+                        endpoint
+                        for member in teams[winner]
+                        if (endpoint := (cast.get(member) or {}).get("model_endpoint"))
+                    ]
+        self.conductor.finalize(
+            reason,
+            winner=winner,
+            winner_kind=winner_kind,
+            winning_model=winning_model,
+            winning_models=winning_models,
+        )
 
     @property
     def cast(self) -> list[AgentManifest]:

tests/test_config.py

@@ -4,6 +4,7 @@ import pytest
 from pydantic import ValidationError
 
 from src.core.config import (
+    CompetitionConfig,
     ModelProfileConfig,
     ModelsConfig,
     ScenarioConfig,
@@ -36,9 +37,7 @@ class TestValidateAgent:
 
 class TestValidateScenario:
     def test_valid_with_goal_and_cast(self):
-        s = validate_scenario(
-            {"name": "w", "goal": "be strange", "default_seed": "seed", "cast": ["a", "b"]}
-        )
+        s = validate_scenario({"name": "w", "goal": "be strange", "default_seed": "seed", "cast": ["a", "b"]})
         assert s.goal == "be strange"
         assert s.cast == ["a", "b"]
 
@@ -65,3 +64,131 @@ class TestValidateWorld:
                 }
             )
         assert "undefined agents" in str(exc.value)
+
+
+# ── competition contract (ADR-0029) ──────────────────────────────────────────────
+
+
+class TestCompetitionConfig:
+    """A scenario's winner contract — versus/judged/none and the team-shape rules
+    a competition can enforce on its own (cross-cast checks live on WorldConfig)."""
+
+    def test_default_is_none_with_no_teams(self):
+        # An absent block == none; the field must default safely, never to versus.
+        c = CompetitionConfig()
+        assert c.kind == "none"
+        assert c.teams is None
+
+    def test_judged_needs_no_teams(self):
+        c = CompetitionConfig(kind="judged")
+        assert c.kind == "judged"
+        assert c.teams is None
+
+    @pytest.mark.parametrize("kind", ["none", "judged"])
+    def test_teams_forbidden_unless_versus(self, kind):
+        # teams on a non-versus kind is a config mistake — the winner has no team map.
+        with pytest.raises(ValidationError) as exc:
+            CompetitionConfig(kind=kind, teams={"spy": ["a"]})
+        assert "only allowed when kind is 'versus'" in str(exc.value)
+
+    def test_versus_requires_non_empty_teams(self):
+        with pytest.raises(ValidationError) as exc:
+            CompetitionConfig(kind="versus", teams={})
+        assert "non-empty 'teams'" in str(exc.value)
+
+    def test_versus_with_missing_teams_rejected(self):
+        # kind=versus with no teams at all is the same defect as an empty mapping.
+        with pytest.raises(ValidationError):
+            CompetitionConfig(kind="versus")
+
+    def test_versus_rejects_empty_member_list(self):
+        # A team with no members can never win or lose — reject it at config time.
+        with pytest.raises(ValidationError) as exc:
+            CompetitionConfig(kind="versus", teams={"spy": ["spy-nil"], "herd": []})
+        assert "empty member lists" in str(exc.value)
+        assert "herd" in str(exc.value)
+
+    def test_versus_rejects_overlapping_teams(self):
+        # An agent on two teams makes "who won" ambiguous — disjointness is required.
+        with pytest.raises(ValidationError) as exc:
+            CompetitionConfig(kind="versus", teams={"spy": ["nil"], "herd": ["cara", "nil"]})
+        assert "mutually disjoint" in str(exc.value)
+        assert "nil" in str(exc.value)
+
+    def test_versus_disjoint_teams_accepted(self):
+        c = CompetitionConfig(kind="versus", teams={"spy": ["nil"], "herd": ["cara", "bex"]})
+        assert c.teams == {"spy": ["nil"], "herd": ["cara", "bex"]}
+
+    def test_same_member_repeated_within_one_team_is_not_overlap(self):
+        # Overlap means across DIFFERENT labels; a dup inside one team is harmless here.
+        c = CompetitionConfig(kind="versus", teams={"spy": ["nil", "nil"]})
+        assert c.kind == "versus"
+
+    def test_extra_field_forbidden(self):
+        with pytest.raises(ValidationError):
+            CompetitionConfig(kind="none", bogus=1)  # type: ignore[call-arg]
+
+
+class TestScenarioCompetition:
+    def test_scenario_accepts_competition_block(self):
+        s = validate_scenario(
+            {
+                "name": "duel",
+                "default_seed": "seed",
+                "cast": ["a", "b"],
+                "competition": {"kind": "versus", "teams": {"x": ["a"], "y": ["b"]}},
+            }
+        )
+        assert s.competition is not None
+        assert s.competition.kind == "versus"
+
+    def test_scenario_without_competition_defaults_to_none_attribute(self):
+        # Absent block == no competition object (the hook reads this as "none").
+        s = validate_scenario({"name": "s", "default_seed": "seed", "cast": ["a"]})
+        assert s.competition is None
+
+
+class TestWorldCompetitionCrossChecks:
+    """WorldConfig is where a team's members are checked against the scenario cast
+    and team labels against agent names — the rules that need the whole world."""
+
+    def _world(self, competition: dict) -> dict:
+        return {
+            "agents": [
+                {"name": "spy-nil", "persona": "p"},
+                {"name": "spy-cara", "persona": "p"},
+                {"name": "host", "persona": "p"},
+            ],
+            "scenarios": [
+                {
+                    "name": "duel",
+                    "default_seed": "seed",
+                    "cast": ["spy-nil", "spy-cara", "host"],
+                    "competition": competition,
+                }
+            ],
+        }
+
+    def test_coherent_versus_world_validates(self):
+        world = validate_world(self._world({"kind": "versus", "teams": {"spy": ["spy-nil"], "herd": ["spy-cara"]}}))
+        assert world.scenarios[0].competition.kind == "versus"
+
+    def test_off_cast_team_member_rejected(self):
+        # A team naming an agent not in this scenario's cast can never be scored.
+        with pytest.raises(ValidationError) as exc:
+            validate_world(self._world({"kind": "versus", "teams": {"spy": ["ghost-agent"], "herd": ["spy-cara"]}}))
+        assert "team members not in its cast" in str(exc.value)
+        assert "ghost-agent" in str(exc.value)
+
+    def test_team_label_colliding_with_agent_name_rejected(self):
+        # winner carries an agent name OR a team label; a label that IS an agent name
+        # makes that union ambiguous, so the cross-cast check must reject it.
+        with pytest.raises(ValidationError) as exc:
+            validate_world(self._world({"kind": "versus", "teams": {"host": ["spy-nil"], "herd": ["spy-cara"]}}))
+        assert "collide with agent names" in str(exc.value)
+        assert "host" in str(exc.value)
+
+    def test_judged_scenario_skips_team_checks(self):
+        # No teams means the cross-cast loop has nothing to enforce — it must pass.
+        world = validate_world(self._world({"kind": "judged"}))
+        assert world.scenarios[0].competition.kind == "judged"

tests/test_run_history.py

@@ -130,6 +130,28 @@ class TestRunFinished:
         assert len(finished) == 1
         assert finished[0].payload["reason"] == "user_stop"
 
+    def test_versus_session_finalizes_with_team_winner(self):
+        # End-to-end offline: the SpyHost stamps a team winner ("herd"/"spy") on the
+        # verdict, and FishbowlSession.finalize must resolve winner_kind == "team"
+        # (a team label, not a cast agent), never guessing a single winning_model.
+        session = FishbowlSession("the-steeped")
+        session.reset()
+        for _ in range(session.autoplay_tick_cap):
+            if session.has_verdict():
+                break
+            session.step()
+        assert session.has_verdict(), "the stub spy-host should reach a verdict"
+
+        session.finalize("verdict")
+        finished = [
+            e for e in session.conductor.ledger.events_for_run(session.conductor.run_id) if e.kind == "run.finished"
+        ]
+        assert len(finished) == 1
+        payload = finished[0].payload
+        assert payload["winner"] in ("herd", "spy")  # a team label, code-stamped
+        assert payload["winner_kind"] == "team"
+        assert payload["winning_model"] is None  # never guessed for a team win
+
     def test_session_finalize_derives_winner_and_model_from_verdict(self):
         scenario = _verdict_scenario()
         # Build a session-like conductor manually so we control the cast verdict.
@@ -302,7 +324,9 @@ def _bookended_runs() -> list[Event]:
             payload={
                 "reason": "verdict",
                 "winner": "judge",
+                "winner_kind": "agent",
                 "winning_model": "model://j",
+                "winning_models": ["model://j"],
                 "turns": 2,
                 "tokens": 1234,
             },
@@ -312,6 +336,42 @@ def _bookended_runs() -> list[Event]:
     ]
 
 
+def _team_win_run() -> list[Event]:
+    """A single versus run finishing on a TEAM win (winner_kind 'team', no single model)."""
+    return [
+        Event(
+            run_id="v1",
+            turn=0,
+            kind="run.started",
+            actor="conductor",
+            payload={
+                "seed": "leaf",
+                "scenario": "the-steeped",
+                "cast": {
+                    "spy-cara": {"model_endpoint": "model://cara", "model_profile": "fast"},
+                    "spy-bex": {"model_endpoint": "model://bex", "model_profile": "fast"},
+                    "spy-ovo": {"model_endpoint": None, "model_profile": "fast"},
+                },
+            },
+        ),
+        Event(
+            run_id="v1",
+            turn=2,
+            kind="run.finished",
+            actor="conductor",
+            payload={
+                "reason": "verdict",
+                "winner": "herd",
+                "winner_kind": "team",
+                "winning_model": None,  # never guessed for a team win
+                "winning_models": ["model://cara", "model://bex"],  # None member endpoint dropped
+                "turns": 2,
+                "tokens": 50,
+            },
+        ),
+    ]
+
+
 class TestRunIndexModule:
     def test_index_runs_folds_bookend_events(self):
         summaries = index_runs(_bookended_runs())
@@ -323,6 +383,9 @@ class TestRunIndexModule:
         assert r1.cast["judge"].model_endpoint == "model://j"
         assert r1.cast["judge"].model_profile == "strong"
         assert (r1.reason, r1.winner, r1.winning_model) == ("verdict", "judge", "model://j")
+        # ADR-0029 attribution keys round-trip through the projection (agent win).
+        assert r1.winner_kind == "agent"
+        assert r1.winning_models == ["model://j"]
         assert (r1.turns, r1.tokens) == (2, 1234)
         assert r1.started_at is not None and r1.finished_at is not None
 
@@ -331,12 +394,26 @@ class TestRunIndexModule:
         assert r2.reason is None and r2.winner is None
         assert (r2.turns, r2.tokens) == (0, 0)
         assert r2.finished_at is None
+        # An unfinished run carries the additive ADR-0029 defaults, never a stale guess.
+        assert r2.winner_kind is None
+        assert r2.winning_models == []
+
+    def test_index_runs_round_trips_a_team_win(self):
+        # winner_kind 'team' carries no single winning_model, only the member endpoints —
+        # and a None member endpoint is dropped from winning_models.
+        (summary,) = index_runs(_team_win_run())
+        assert summary.winner == "herd"
+        assert summary.winner_kind == "team"
+        assert summary.winning_model is None
+        assert summary.winning_models == ["model://cara", "model://bex"]
 
     @pytest.mark.parametrize("make_ledger", [Ledger, lambda: SqlAlchemyLedger("sqlite://")])
-    def test_from_ledger_matches_pure_oracle(self, make_ledger):
-        events = _bookended_runs()
+    @pytest.mark.parametrize("events_factory", [_bookended_runs, _team_win_run])
+    def test_from_ledger_matches_pure_oracle(self, make_ledger, events_factory):
+        events = events_factory()
         ledger = make_ledger()
         for e in events:
             ledger.append(e)
-        # The indexed-query path must produce exactly what the pure oracle does.
+        # The indexed-query path must produce exactly what the pure oracle does —
+        # including the additive ADR-0029 attribution keys (agent win and team win).
         assert index_runs_from_ledger(ledger) == index_runs(events)

tests/test_structured.py

@@ -1,6 +1,11 @@
 from __future__ import annotations
 
+import pytest
+from pydantic import ValidationError
+
 from src.core.structured import (
+    AgentOutputError,
+    build_output_model,
     clean_clue,
     extract_reasoning,
     is_usable_line,
@@ -198,3 +203,85 @@ class TestIsUsableLine:
 
     def test_accepts_a_real_line(self):
         assert is_usable_line("A dark brew warms the dawn.")
+
+
+class TestBuildOutputModel:
+    """The live-path schema. kind is Literal-constrained, text is required, and the
+    well-known verdict fields (ADR-0029) get real optional types — everything else
+    stays a required string, exactly as before."""
+
+    def test_requires_at_least_one_kind(self):
+        with pytest.raises(AgentOutputError):
+            build_output_model([])
+
+    def test_kind_is_literal_constrained(self):
+        model = build_output_model(["judge.verdict"])
+        with pytest.raises(ValidationError):
+            model(kind="not.allowed", text="x")
+
+    def test_text_is_required(self):
+        model = build_output_model(["agent.spoke"])
+        with pytest.raises(ValidationError):
+            model(kind="agent.spoke")
+
+    def test_ordinary_extra_field_is_required_string(self):
+        # The *other* row: an arbitrary scenario field stays a required str (back-compat).
+        model = build_output_model(["agent.spoke"], extra_fields=["mood"])
+        with pytest.raises(ValidationError):
+            model(kind="agent.spoke", text="hi")  # mood missing → invalid
+        inst = model(kind="agent.spoke", text="hi", mood="calm")
+        assert inst.mood == "calm"
+
+    def test_winner_is_optional_and_defaults_to_none(self):
+        # A judge may decline to name a winner; the field must default to None, not error.
+        model = build_output_model(["judge.verdict"], extra_fields=["winner"])
+        inst = model(kind="judge.verdict", text="Verdict: undecided.")
+        assert inst.winner is None
+
+    def test_winner_accepts_a_name(self):
+        model = build_output_model(["judge.verdict"], extra_fields=["winner"])
+        inst = model(kind="judge.verdict", text="t", winner="clue-gatherer")
+        assert inst.winner == "clue-gatherer"
+
+    def test_scores_defaults_to_empty_map(self):
+        model = build_output_model(["judge.verdict"], extra_fields=["scores"])
+        inst = model(kind="judge.verdict", text="t")
+        assert inst.scores == {}
+
+    def test_scores_coerces_numeric_map(self):
+        model = build_output_model(["judge.verdict"], extra_fields=["scores"])
+        inst = model(kind="judge.verdict", text="t", scores={"clue-gatherer": 9})
+        assert inst.scores == {"clue-gatherer": 9.0}
+
+    def test_mixed_known_and_unknown_fields(self):
+        # mood required, winner/scores optional — the full mystery-judge shape.
+        model = build_output_model(["judge.verdict"], extra_fields=["mood", "winner", "scores"])
+        inst = model(kind="judge.verdict", text="t", mood="smug")
+        assert (inst.mood, inst.winner, inst.scores) == ("smug", None, {})
+
+
+class TestJsonInstructionWellKnown:
+    """The prompt hint. With NO well-known field present it must be byte-identical to
+    the original uniform schema; with winner/scores present it renders typed hints."""
+
+    @pytest.mark.parametrize("extra", [None, ["mood"], ["thought"], ["mood", "thought"]])
+    def test_byte_identical_without_well_known_fields(self, extra):
+        # The common case must not drift: the same schema-line rendering as before.
+        out = json_instruction(["agent.spoke"], extra_fields=extra)
+        fields = '", "'.join(["kind", "text", *(extra or [])])
+        assert f'Schema: {{"{fields}": "..."}}' in out
+
+    def test_winner_hint_appears_when_present(self):
+        out = json_instruction(["judge.verdict"], extra_fields=["winner"])
+        assert "winner" in out
+        assert "or null" in out  # the typed hint, not the uniform "..."
+
+    def test_scores_hint_appears_when_present(self):
+        out = json_instruction(["judge.verdict"], extra_fields=["scores"])
+        assert "0-10" in out  # a number-map hint, not a quoted string
+
+    def test_ordinary_field_keeps_uniform_hint_alongside_known(self):
+        # mood sits next to winner: it still gets "..." while winner gets its typed hint.
+        out = json_instruction(["judge.verdict"], extra_fields=["mood", "winner", "scores"])
+        assert '"mood": "..."' in out
+        assert "or null" in out and "0-10" in out

tests/test_verdict_validation.py

@@ -0,0 +1,370 @@
+"""Verdict winner validation and the ground-truth scoreboard (ADR-0029).
+
+The base agent validates a judge's ``winner`` (re-asking once on a bad pick, summing
+usage, stamping ``no_contest`` on a second failure) and normalises ``scores`` in place.
+The ``SpyHost`` handler then turns the judge's *accusation* into a code-stamped
+*result* using the scenario's ``competition.teams``.
+
+Zero mocks: agents are built through the real registry, and the live ``complete_structured``
+seam is exercised with a small hand-written ``FakeProvider`` (the canonical zero-mock
+seam) and an offline ``DeterministicTinyModel`` for the stub path.
+"""
+
+from __future__ import annotations
+
+from src.agents.handlers import SpyHost
+from src.core.events import Event
+from src.core.projections import StageProjection
+from src.core.registry import default_registry
+from src.models.router import ModelRouter
+
+
+# ── live-path provider seam (no unittest.mock) ────────────────────────────────────
+
+
+class _ScriptedJudge:
+    """A live provider whose ``complete_structured`` returns scripted verdicts.
+
+    Exposing ``complete_structured`` is what routes the base agent down the live
+    path (``hasattr(provider, "complete_structured")``).  Each call returns the next
+    scripted ``(winner, scores)`` and reports the matching usage, so a re-ask is a
+    real second round-trip the test can count and meter."""
+
+    def __init__(self, scripts: list[dict]) -> None:
+        self._scripts = scripts
+        self.calls = 0
+        self.last_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
+        self.last_reasoning = ""
+        self.model_id = "scripted"
+
+    def complete_structured(self, role, prompt, model):
+        script = self._scripts[min(self.calls, len(self._scripts) - 1)]
+        self.calls += 1
+        usage = script.get("usage", {"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110})
+        self.last_usage = dict(usage)
+        return model(
+            kind=script.get("kind", "judge.verdict"),
+            text=script.get("text", "Verdict: someone did it."),
+            mood=script.get("mood", "calm"),
+            winner=script.get("winner"),
+            scores=script.get("scores", {}),
+        )
+
+
+class _ScriptedRouter(ModelRouter):
+    """A router that always hands back one scripted provider (the live seam)."""
+
+    def __init__(self, provider) -> None:
+        self._provider = provider
+        self.offline = False
+
+    def for_profile(self, key):  # type: ignore[override]
+        return self._provider
+
+
+def _judge(scripts: list[dict], *, scenario: str = "mystery-roots", name: str = "mystery-judge"):
+    """Build a judge through the registry, wired to a scripted live provider."""
+    reg = default_registry()
+    provider = _ScriptedJudge(scripts)
+    agent = reg.build_agent(name, _ScriptedRouter(provider))
+    cfg = reg.scenarios[scenario]
+    agent.competition = cfg.competition
+    agent.cast_names = list(cfg.cast)
+    return agent, provider
+
+
+def _resolve(agent):
+    return agent._resolve_payload(
+        agent.manifest.name,
+        "PROMPT",
+        agent._content_kinds(),
+        agent.manifest.output_extra_fields,
+    )
+
+
+# ── live re-ask flow ───────────────────────────────────────────────────────────────
+
+
+class TestVerdictReask:
+    def test_off_cast_winner_triggers_exactly_one_reask(self):
+        # The reference flow: bad winner first, valid winner on the corrective re-ask.
+        agent, provider = _judge(
+            [
+                {
+                    "text": "Verdict: the butler did it.",
+                    "winner": "NOT-A-CAST-NAME",
+                    "scores": {"clue-gatherer": 12},
+                    "usage": {"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110},
+                },
+                {
+                    "text": "Verdict: the gatherer cracked it.",
+                    "winner": "clue-gatherer",
+                    "scores": {"clue-gatherer": 9},
+                    "usage": {"prompt_tokens": 120, "completion_tokens": 12, "total_tokens": 132},
+                },
+            ]
+        )
+        payload = _resolve(agent)
+        assert provider.calls == 2  # one re-ask, not two, not zero
+        assert payload["winner"] == "clue-gatherer"
+        assert "no_contest" not in payload
+
+    def test_reask_usage_is_summed_not_overwritten(self):
+        # ADR-0029 acceptance criterion: the governor must meter BOTH calls.
+        agent, _ = _judge(
+            [
+                {"winner": "bad", "usage": {"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110}},
+                {
+                    "winner": "clue-gatherer",
+                    "text": "Verdict: the gatherer cracked it.",
+                    "usage": {"prompt_tokens": 120, "completion_tokens": 12, "total_tokens": 132},
+                },
+            ]
+        )
+        _resolve(agent)
+        assert agent.last_usage["total_tokens"] == 242  # 110 + 132
+        assert agent.last_usage["prompt_tokens"] == 220
+        assert agent.last_usage["completion_tokens"] == 22
+
+    def test_valid_first_try_does_not_reask(self):
+        agent, provider = _judge([{"winner": "hypothesis-former", "text": "Verdict: the hypothesis held."}])
+        payload = _resolve(agent)
+        assert provider.calls == 1
+        assert payload["winner"] == "hypothesis-former"
+        assert "no_contest" not in payload
+
+    def test_reask_that_also_fails_drops_winner_and_stamps_no_contest(self):
+        # Two bad picks in a row: the verdict TEXT still ships, the row is forfeited.
+        agent, provider = _judge(
+            [
+                {"winner": "bad-one", "text": "Verdict: I accuse the wind."},
+                {"winner": "bad-two", "text": "Verdict: no, the rain."},
+            ]
+        )
+        payload = _resolve(agent)
+        assert provider.calls == 2
+        assert payload.get("no_contest") is True
+        assert "winner" not in payload
+        assert payload["text"]  # the drama survives — the show always ends
+
+    def test_missing_winner_is_not_an_error(self):
+        # winner is optional; a judge that names none must NOT trigger a re-ask.
+        agent, provider = _judge([{"winner": None, "text": "Verdict: the evidence is mute."}])
+        payload = _resolve(agent)
+        assert provider.calls == 1
+        assert payload.get("winner") is None
+        assert "no_contest" not in payload
+
+
+class TestScoresNormalisation:
+    """scores is garnish: cleaned in place, never re-asked."""
+
+    def test_non_cast_keys_dropped_and_values_clamped(self):
+        agent, provider = _judge(
+            [
+                {
+                    "winner": "clue-gatherer",
+                    "text": "Verdict: the gatherer cracked it.",
+                    "scores": {"clue-gatherer": 12, "ghost": 5, "hypothesis-former": -3},
+                }
+            ]
+        )
+        payload = _resolve(agent)
+        assert provider.calls == 1  # scores never cause a re-ask
+        # ghost is off-cast → dropped; 12 → clamped to 10; -3 → clamped to 0.
+        assert payload["scores"] == {"clue-gatherer": 10.0, "hypothesis-former": 0.0}
+
+    def test_bad_winner_with_scores_still_only_reasks_for_winner(self):
+        agent, provider = _judge(
+            [
+                {"winner": "bad", "scores": {"clue-gatherer": 99}},
+                {
+                    "winner": "clue-gatherer",
+                    "text": "Verdict: the gatherer cracked it.",
+                    "scores": {"clue-gatherer": 7},
+                },
+            ]
+        )
+        payload = _resolve(agent)
+        assert provider.calls == 2
+        assert payload["scores"] == {"clue-gatherer": 7.0}
+
+
+class TestHookInert:
+    """The validation hook is a transparent pass-through for everything that is not a
+    judge in a live competition — even a downright odd winner slips through untouched."""
+
+    def test_judged_winner_passes_unchanged(self):
+        agent, provider = _judge([{"winner": "devils-advocate", "text": "Verdict: the advocate won."}])
+        payload = _resolve(agent)
+        assert provider.calls == 1
+        assert payload["winner"] == "devils-advocate"
+
+    def test_non_judge_role_leaves_odd_winner_untouched(self):
+        # The hook keys on role == "judge".  Take the judge's exact schema (so the
+        # winner field and verdict kind stay valid) but flip the role to a worker: the
+        # hook must go inert, so an off-cast winner rides through verbatim — no
+        # validation, no re-ask, no no_contest.
+        reg = default_registry()
+        provider = _ScriptedJudge([{"winner": "anything-goes", "text": "Verdict-shaped, but a worker said it."}])
+        agent = reg.build_agent("mystery-judge", _ScriptedRouter(provider))
+        cfg = reg.scenarios["mystery-roots"]
+        agent.competition = cfg.competition
+        agent.cast_names = list(cfg.cast)
+        agent.manifest = agent.manifest.model_copy(update={"role": "worker"})
+        payload = _resolve(agent)
+        assert provider.calls == 1  # no re-ask despite the off-cast winner
+        assert payload["winner"] == "anything-goes"
+        assert "no_contest" not in payload
+
+    def test_no_competition_attached_leaves_winner_untouched(self):
+        # A bare-built judge (no registry injection) has competition=None → hook inert.
+        reg = default_registry()
+        provider = _ScriptedJudge([{"winner": "off-the-wall", "text": "Verdict: chaos reigns."}])
+        agent = reg.build_agent("mystery-judge", _ScriptedRouter(provider))
+        # Deliberately do NOT set agent.competition / cast_names.
+        assert agent.competition is None
+        payload = _resolve(agent)
+        assert provider.calls == 1
+        assert payload["winner"] == "off-the-wall"
+
+    def test_none_kind_competition_is_inert(self):
+        from src.core.config import CompetitionConfig
+
+        reg = default_registry()
+        provider = _ScriptedJudge([{"winner": "whoever", "text": "Verdict: nobody wins the wood."}])
+        agent = reg.build_agent("mystery-judge", _ScriptedRouter(provider))
+        agent.competition = CompetitionConfig(kind="none")
+        agent.cast_names = ["clue-gatherer"]
+        payload = _resolve(agent)
+        assert provider.calls == 1
+        assert payload["winner"] == "whoever"
+
+
+# ── offline path: re-ask works there too, and the stub never triggers it ───────────
+
+
+class TestOfflineVerdictPath:
+    def test_offline_judge_emits_clean_winnerless_payload_no_reask(self):
+        # The stub's _synth_field returns None/​{} for winner/scores, so an offline
+        # judge produces a validation-clean payload with no wasted corrective round-trip.
+        reg = default_registry()
+        agent = reg.build_agent("mystery-judge", ModelRouter(offline=True))
+        cfg = reg.scenarios["mystery-roots"]
+        agent.competition = cfg.competition
+        agent.cast_names = list(cfg.cast)
+        payload = _resolve(agent)
+        assert payload["winner"] is None
+        assert payload["scores"] == {}
+        assert "no_contest" not in payload
+
+
+# ── SpyHost ground-truth scoreboard ────────────────────────────────────────────────
+
+
+def _spy_host() -> SpyHost:
+    reg = default_registry()
+    host = reg.build_agent("spy-host", ModelRouter(offline=True))
+    cfg = reg.scenarios["the-steeped"]
+    host.competition = cfg.competition
+    host.cast_names = list(cfg.cast)
+    return host
+
+
+class TestScanAccusation:
+    """_scan_accusation recovers the accused from verdict text by the distinctive tail
+    of each cast name (``spy-cara`` → ``cara``), earliest mention wins, host excluded."""
+
+    def test_finds_named_player(self):
+        host = _spy_host()
+        assert host._scan_accusation("Verdict: I point at NIL.") == "spy-nil"
+
+    def test_matches_case_insensitively(self):
+        host = _spy_host()
+        assert host._scan_accusation("CARA slipped a tell at dawn.") == "spy-cara"
+
+    def test_earliest_mention_wins(self):
+        host = _spy_host()
+        # BEX appears before NIL → bex is the accusation, even though both are named.
+        assert host._scan_accusation("BEX hesitated, then NIL steeped too fast.") == "spy-bex"
+
+    def test_host_token_never_self_accuses(self):
+        host = _spy_host()
+        # The host's own name token ("host") is excluded; no player named → None.
+        assert host._scan_accusation("The host weighs the room and the clues.") is None
+
+    def test_no_recoverable_name_returns_none(self):
+        host = _spy_host()
+        assert host._scan_accusation("A long deliberation with no names at all.") is None
+
+
+class TestStampScoreboard:
+    """The load-bearing split: code turns the judge's accusation into the result."""
+
+    def _verdict_event(self, **payload) -> Event:
+        return Event(run_id="r", turn=1, kind="judge.verdict", actor="spy-host", payload=payload)
+
+    def test_correct_accusation_lets_the_herd_win(self):
+        host = _spy_host()
+        event = self._verdict_event(text="Verdict: NIL is the spy.", winner="spy-nil")
+        host._stamp_scoreboard(event)
+        assert event.payload["accused"] == "spy-nil"
+        assert event.payload["correct"] is True
+        assert event.payload["winner"] == "herd"  # spy caught → herd wins
+
+    def test_wrong_accusation_lets_the_spy_win(self):
+        host = _spy_host()
+        event = self._verdict_event(text="Verdict: CARA is the spy.", winner="spy-cara")
+        host._stamp_scoreboard(event)
+        assert event.payload["accused"] == "spy-cara"
+        assert event.payload["correct"] is False
+        assert event.payload["winner"] == "spy"  # innocent accused → spy wins
+
+    def test_offline_accusation_recovered_from_text(self):
+        # No winner field (the offline shape) → the accusation is scanned from text.
+        host = _spy_host()
+        event = self._verdict_event(text="Verdict: I point at NIL. The herd's clues brewed.")
+        host._stamp_scoreboard(event)
+        assert event.payload["accused"] == "spy-nil"
+        assert event.payload["winner"] == "herd"
+
+    def test_no_recoverable_accusation_is_no_contest(self):
+        host = _spy_host()
+        event = self._verdict_event(text="A long deliberation with no names at all.")
+        host._stamp_scoreboard(event)
+        assert event.payload.get("no_contest") is True
+        assert "winner" not in event.payload
+
+    def test_no_spy_team_is_inert(self):
+        # A versus competition without a 'spy' team has no ground truth to stamp.
+        from src.core.config import CompetitionConfig
+
+        host = _spy_host()
+        host.competition = CompetitionConfig(kind="versus", teams={"a": ["spy-cara"], "b": ["spy-bex"]})
+        event = self._verdict_event(text="Verdict: NIL.", winner="spy-nil")
+        host._stamp_scoreboard(event)
+        # Untouched: no accused/correct/no_contest stamped, winner left as the raw pick.
+        assert "accused" not in event.payload
+        assert "correct" not in event.payload
+        assert event.payload["winner"] == "spy-nil"
+
+
+class TestSpyHostActEnrichment:
+    """The full handler turn: super().act() produces the verdict, then the scoreboard
+    is stamped — driven entirely offline through the deterministic stub."""
+
+    def test_act_stamps_a_full_scoreboard_offline(self):
+        host = _spy_host()
+        # The stub's spy-host lines all name NIL → a deterministic herd win.
+        recent = tuple(
+            Event(run_id="r", turn=1, kind="agent.spoke", actor=p, payload={"text": "a clue"})
+            for p in ("spy-cara", "spy-bex", "spy-nil", "spy-ovo")
+        )
+        projection = StageProjection()
+        event = host.act("r", 2, projection, recent)
+        assert event.kind == "judge.verdict"
+        assert event.payload["accused"] == "spy-nil"
+        assert event.payload["correct"] is True
+        assert event.payload["winner"] == "herd"
+        # The dramatic reveal still rides alongside the scoreboard.
+        assert isinstance(event.payload.get("reveal"), list) and event.payload["reveal"]