feat: implement Archive feature for past session management

- Added `archive.py` to handle listing and loading past runs for a specific user and scenario.
- Introduced `list_runs` function to retrieve past runs scoped by user session ID and scenario.
- Implemented `load_replay` function to create a read-only `ReplaySession` for a given run ID.
- Created helper functions for formatting run cards and timestamps.
- Enhanced CSS styles for the Archive drawer and run cards to improve UI presentation.
- Updated `FishbowlSession` and `ReplaySession` classes to support new functionality and ensure session isolation.
- Added comprehensive tests for the Archive feature, ensuring correct session attribution and replay functionality.
- Modified memory index tests to include run ID scoping for event searches.
- Created new tests for run history and lifecycle management, ensuring proper event emission and session handling.

docs/adr/0026-run-lifecycle-and-history.md

@@ -0,0 +1,135 @@
+# ADR-0026: Run Lifecycle, Provenance, and Multi-Run History in One Ledger
+
+## Status
+
+Accepted
+
+## Context
+
+The engine has always known how a run *begins* — `Conductor.reset()` mints a `run_id`
+and appends `run.started` — but it never recorded how a run *ends*, and `reset()` itself
+was destructive in a way that fights the persistent store. Three gaps:
+
+1. **No terminal event.** A run stopped for many reasons — the judge reached a
+   `judge.verdict`, the governor tripped (`BudgetExceeded`, ADR-0013), the tick cap was
+   hit, or the operator pressed stop — but the ledger held no record of *which*, nor a
+   tidy summary (who won, on what model, how many turns/tokens). Every consumer (UI,
+   trace export, resume) had to re-derive "is this run over and why" from the tail.
+2. **`reset()` wiped the database.** `reset()` called `self.ledger.reset()`, which on a
+   persistent backend (`make_ledger()` always returns `SqlAlchemyLedger`; the Fishbowl
+   session already uses it) erased *every* prior run. Starting a second show silently
+   destroyed the first — directly at odds with wanting a durable, shareable history.
+3. **Thin provenance + run-blind projections.** `run.started` carried only `{seed, goal}`,
+   so a replay could not tell which scenario ran or which model each agent was bound to
+   (ADR-0022's per-agent binding was invisible after the fact). And projections/memory
+   rebuilt from *all* ledger events, so they could not be scoped to a single run once the
+   DB held many.
+
+The hackathon bar wants the append-only ledger to *be* the shareable agent trace
+(Sharing-is-Caring badge). A trace you can't bound to one run, that loses its own start
+provenance, and that gets clobbered on the next Summon is not shareable.
+
+## Decision
+
+Treat the ledger as a **multi-run history**: one durable store holds many runs, each
+framed by an enriched `run.started` and a new `run.finished`, queryable per run. All
+changes are additive — `schema_version` stays **1** (ADR-0009 open/additive kinds), no
+migration.
+
+- **New terminal kind `run.finished`** (added to `CORE_EVENT_KINDS`, regex-valid under
+  ADR-0009). Payload shape:
+
+  ```
+  {
+    "reason": "verdict" | "budget" | "tick_cap" | "user_stop",
+    "winner": str | None,          # actor name of the winner, if any
+    "winning_model": str | None,   # model_endpoint bound to the winner, if known
+    "turns": int,                  # turns elapsed in the run
+    "tokens": int                  # total tokens spent in the run
+  }
+  ```
+
+  `winner`/`winning_model` are populated from a `judge.verdict` when the reason is
+  `verdict` (the winner's model is resolved through the cast→model map below); both are
+  `null` for budget/tick-cap/user-stop endings. `turns`/`tokens` are read from the
+  governor's live stats (`src/core/governor.py`).
+
+- **Enriched `run.started` payload.** Keep the existing `{seed, goal}` and add, alongside
+  them, the **scenario name** and a **cast→model binding map** (ADR-0022 made visible
+  *after* the fact):
+
+  ```
+  {
+    "seed": str, "goal": str,                       # existing — unchanged
+    "scenario": str,                                # scenario name
+    "cast": {                                       # one entry per agent
+       "<agent_name>": {"model_endpoint": str | None,
+                        "model_profile": "tiny"|"fast"|"balanced"|"strong"}
+    }
+  }
+  ```
+
+  The map is built from `self.scenario.agents`, reading each `agent.manifest.model_endpoint`
+  / `model_profile` and guarding agents that lack a manifest (`getattr(agent,"manifest",None)`).
+  Purely additive keys, so old readers ignore them and `schema_version` stays 1.
+
+- **Non-destructive `reset()`.** On a persistent ledger, `reset()` **no longer wipes the
+  DB**. It mints a new `run_id`, clears only the conductor's *in-memory* turn state
+  (queues, `agent_errors`, governor, turn counter), and appends a fresh enriched
+  `run.started` plus the scenario genesis. Prior runs stay on disk, so one DB accumulates
+  a history of shows. (The in-memory `Ledger` used by some tests is naturally fresh per
+  instance; the destructive `ledger.reset()` call is dropped from the lifecycle.)
+
+- **Run-scoped projections (optional `run_id`).** `rebuild_stage(events, run_id=None)` and
+  `EpisodicMemory/SalienceMemory.visible(events, run_id=None)` gain an optional `run_id`
+  that, when given, filters the tuple to that run before projecting — trivial because
+  every `Event` carries `run_id`. Default `None` preserves today's whole-ledger behaviour,
+  so existing callers and tests are untouched.
+
+- **New ledger query API.** Add `events_for_run(run_id) -> tuple[Event, ...]` and
+  `runs() -> list[...]` (one descriptor per `run.started`, newest first, with id + scenario
+  + start time) to the ledger surface. Both SQL backends already index `run_id`, so
+  `events_for_run` is an indexed query mirroring the existing `tail(from_offset)` /
+  `latest_offset()` style; the in-memory backend filters its `events` tuple.
+
+## Consequences
+
+- **The trace is self-describing and bounded.** A single run.started→…→run.finished slice
+  names its scenario, its full cast→model binding, how it ended, who won on what model, and
+  its turn/token cost — ready to export to the HF Hub as a stand-alone trace.
+- **History survives.** Summoning a new show no longer destroys prior runs; one DB is a
+  gallery of shows. The UI can list `runs()` and replay any one via `events_for_run`.
+- **Run-scoped views, no double counting.** Stage and memory can be rebuilt for a chosen
+  run even when the store holds many, so a multi-run DB doesn't bleed one show's lines into
+  another's blackboard.
+- **Governor stops are recorded, not just raised.** `BudgetExceeded` (ADR-0013) still
+  bubbles, but the lifecycle now also writes a `run.finished{reason:"budget"}` so the stop
+  is visible in the trace rather than inferred from an absent next event.
+- **Additive only.** No schema bump, no migration; old ledgers and old readers keep
+  working, and the deterministic offline stub stays byte-reproducible.
+- **Garbage collection is out of scope.** An ever-growing single DB will eventually want
+  pruning/archival of old runs; deferred until the history is large enough to matter.
+
+## Alternatives considered
+
+- **Keep wiping on reset, one run per DB.** Simplest, but loses all history on Summon and
+  makes a shareable multi-run trace impossible — the exact problem this ADR fixes.
+- **Separate DB file per run.** Preserves isolation, but fragments the history, complicates
+  `runs()`/replay, and breaks the "one append-only ledger is the trace" story. Rejected for
+  a single run-scoped store.
+- **Derive end-reason/winner at read time instead of a `run.finished` event.** Keeps the
+  ledger thinner, but every consumer re-implements the heuristic and a budget/user stop
+  leaves no positive marker. An explicit terminal event is the append-only-friendly record.
+- **Bump `schema_version` for the richer payloads.** Unnecessary — all additions are new
+  keys/kinds, which ADR-0009's open/additive contract already covers at version 1.
+
+## References
+
+- Builds on ADR-0009 (open/additive event kinds — no migration for new kinds/keys) and
+  ADR-0022 (per-agent explicit model binding — the source of the cast→model map).
+- Relates to ADR-0013 (token governor — supplies the budget stop reason and turn/token
+  stats) and ADR-0024 (observability).
+- `src/core/events.py` — `CORE_EVENT_KINDS` (`run.finished`), documented payload shapes
+- `src/core/conductor.py` — non-destructive `reset()`, enriched `run.started`, `run.finished`
+- `src/core/projections.py`, `src/core/memory.py` — optional `run_id` scoping
+- `src/core/ledger.py`, `src/core/sqlalchemy_ledger.py` — `events_for_run`, `runs`

docs/adr/0027-per-user-sessions-and-archive-replay.md

@@ -0,0 +1,77 @@
+# ADR-0027: Per-User Sessions and Read-Only Archive Replay
+
+## Status
+
+Accepted
+
+## Context
+
+ADR-0026 made the ledger a durable, multi-run history: many runs live in one store,
+each framed by an enriched `run.started` and a `run.finished`, and the live Conductor
+projection is run-scoped. That unlocked "keep the history of runs" at the data layer —
+but the Fishbowl UI had two gaps that surfaced as soon as the store became shared:
+
+1. **The live transcript bled across runs and scenarios.** `FishbowlSession.events`
+   returned `conductor.ledger.events` — the *entire* shared ledger — and `snapshot()`
+   fed that unscoped into `view_model_at` → `rebuild_stage`. With one durable store
+   holding every run, the Show for scenario B would replay scenario A's discussion.
+   This is the "show the right discussion for the right session/scenario" bug.
+2. **No notion of "whose run".** Runs were not attributed to a user/browser, so there
+   was no basis for "show me *my* past sessions" — and no UI to load them. Summon
+   already started a fresh run (ADR-0026's non-destructive reset), but there was no way
+   back to a previous one.
+
+The product ask: Summon always starts a fresh conversation; previous sessions appear
+only behind a **Load** affordance; each user/browser gets a unique session id; and the
+Archive lists *my* sessions for *this* world only.
+
+## Decision
+
+Attribute runs to a per-browser session id, scope every Show read to a single run, and
+add a read-only **Archive** that replays a past run without spending tokens. All
+additive; `schema_version` stays **1** (ADR-0009).
+
+- **`run.started.session_id` (optional).** `Conductor.reset(seed, *, session_id=None)`
+  stamps the id onto `run.started` when present (key omitted otherwise). `RunSummary`
+  (ADR-0026) carries `session_id` so the run-index folds it for free. No side table —
+  the run list stays a projection of the log (ADR-0014).
+
+- **The session id is the browser's, persisted in `localStorage`.** Resolved (or minted
+  with `crypto.randomUUID`) by a small JS function on `demo.load`, written into a hidden
+  carrier, and read by Python. It survives reloads, so "my sessions" is stable per user
+  without server-side accounts.
+
+- **Run-scoped live transcript.** `FishbowlSession.events` now returns
+  `events_for_run(conductor.run_id)`. Every downstream read (`head`, `snapshot`, the
+  scrubber, autoplay) flows from this, so scoping one property scopes the whole Show —
+  scenario B can never show scenario A's discussion.
+
+- **`ReplaySession` — a read-only view over one past run.** It exposes the exact read
+  surface the Show renders (`events`/`head`/`snapshot`/`has_verdict`) over a fixed
+  `events_for_run(run_id)` slice, rebuilding cast/meters from the run's own scenario
+  (named on `run.started`). `replay = True`; `step`/`step_one`/`inject` are no-ops, and
+  the autoplay loop replays the recorded prefix then stops — **a load never generates.**
+
+- **`src/ui/fishbowl/archive.py`.** `list_runs(scenario, session_id)` folds the ledger
+  via `index_runs_from_ledger` and keeps only this user's runs in this world, newest
+  first; `load_replay(run_id)` builds the `ReplaySession`; `run_card_label` renders a
+  one-line phosphor card.
+
+- **Archive drawer in the Lab.** A `gr.Accordion` under Summon with a `gr.render` keyed
+  on (scenario, session id) that lists clickable cards; a card loads the replay and
+  jumps to the Show. The render re-runs on world change, when the id resolves from
+  localStorage, and on manual refresh.
+
+## Consequences
+
+- The Show is correct under a shared, multi-user store: each user drives their own
+  `run_id` in their own `gr.State`, sees only their own live run, and can browse only
+  their own history. Multiple browsers naturally share the store while staying isolated.
+- The Archive is "nearly free" on top of ADR-0026 — pure reads, no new persistence.
+- `ReplaySession` cannot resume a past run (read-only by design); resuming would need a
+  live Conductor rehydrated at that `run_id` (a future step if wanted).
+- localStorage ties "my sessions" to a browser profile; clearing storage starts a new
+  identity. Acceptable for a no-accounts demo; a future signed-in id could supersede it.
+- Postgres is unchanged — no new tables — keeping the ledger the single source of truth
+  (ADR-0014). A materialized run-summary table remains an option if listing ever needs
+  to scale beyond folding the log.

docs/adr/0028-session-stamped-events-and-scoped-recall.md

@@ -0,0 +1,61 @@
+# ADR-0028: Session-Stamped Events and Run-Scoped Memory Recall
+
+## Status
+
+Accepted
+
+## Context
+
+ADR-0027 attributed *runs* to a browser session via `run.started.payload.session_id`,
+and scoped the **UI** to one run. Three gaps remained on the data/context side:
+
+1. **Actions were not attributable.** Only `run.started` carried the session id; the
+   events an agent actually produced (`agent.spoke`, `judge.verdict`, …) did not, so
+   "all actions by user X" required a join through `run.started` in every consumer
+   (SQL, mem0, exports).
+2. **Agent context bled across runs.** `Conductor._run_agent` passed the *whole*
+   ledger (`ledger.events` — every run, every user) as `recent_events`, so episodic
+   and salience memory could recall another show's — another user's — discussion
+   into a prompt.
+3. **Semantic recall was unscoped.** The mem0 index (ADR-0018/0019/0020) stored and
+   searched one global namespace (`user_id="ledger"`), ignoring mem0's native
+   `run_id`/`agent_id` identity scopes; cross-run hits crowded the relevance budget.
+
+Additionally, the session id originates in `localStorage` — untrusted client input —
+and reached the ledger unvalidated.
+
+## Decision
+
+- **`Event.session_id: str | None` on the envelope.** Stamped by the Conductor at
+  the single `_append` chokepoint from the run's normalized session id; agents and
+  scenarios never know sessions exist. Nullable (headless runs stay `None`).
+  Persisted as an indexed nullable column in both SQL backends. Additive —
+  `schema_version` stays 1. *No migration shipped:* we are pre-release; recreate dev
+  DB files instead.
+- **Normalize at the engine boundary.** `normalize_session_id()` (events.py) accepts
+  `[A-Za-z0-9._-]{1,64}` and degrades anything else to `None` (logged) — applied in
+  `Conductor.reset` (write path) and `archive.list_runs` (read path). A tampered
+  localStorage can never break Summon or inject garbage into the store.
+- **Run-scoped agent context.** `_run_agent` now passes
+  `events_for_run(self.run_id)` as `recent_events`. One line; closes the prompt
+  bleed for both memory layers, since every recall folds from that slice.
+- **mem0 native scoping.** `MemoryIndex.search(query, k, run_id=None)`; backends
+  store with mem0's native `run_id=event.run_id` / `agent_id=event.actor` and filter
+  search by `run_id`, with a defensive post-filter on the reconstructed event.
+  `SalienceMemory` derives the scope from its candidates (single-run slice → scoped
+  search, free for callers). `session_id` rides in index metadata for forensics.
+- **RunIndex prefers the envelope.** `RunSummary.session_id` folds from
+  `event.session_id` first, payload copy second.
+
+## Consequences
+
+- Every action is directly filterable by session in SQL (`WHERE session_id = ?`,
+  indexed), in mem0 metadata, and in exported traces (the JSONL dump inherits the
+  envelope field for free).
+- Prompts are hermetic per run: neither the episodic window, nor salience ranking,
+  nor semantic recall can surface another run's text. Verified by probe-agent and
+  scoped-search tests.
+- Existing dev databases predate the `session_id` column and must be deleted (or
+  recreated) — accepted in lieu of a migration while pre-release.
+- mem0's `_indexed` dedup set is process-local; a persistent vector store may
+  re-embed after restart (idempotent by `event.id`, so correct — just re-work).

src/core/conductor.py

@@ -39,7 +39,7 @@ from typing import TYPE_CHECKING
 from uuid import uuid4
 
 from src import observability as obs
-from src.core.events import Event
+from src.core.events import Event, normalize_session_id
 from src.core.governor import BudgetExceeded, Governor
 from src.core.ledger import Ledger
 from src.core.projections import StageProjection, rebuild_stage
@@ -69,6 +69,9 @@ class Conductor:
         self.snapshot_every = snapshot_every
         self.snapshot_path = snapshot_path
         self.run_id = str(uuid4())
+        # The browser/user session driving the current run (normalized, untrusted
+        # input) — stamped onto every event this conductor appends (see _append).
+        self.session_id: str | None = None
         self.turn = 0
         self._trigger_queue: deque[tuple["Agent", Event]] = deque()
         # Actors still to act in the CURRENT turn — the queue ``step_one`` drains one
@@ -84,33 +87,109 @@ class Conductor:
 
     @property
     def projection(self) -> StageProjection:
-        return rebuild_stage(self.ledger.events)
+        # Run-scoped: the live stage shows only the current run, even though the
+        # ledger is a shared, append-only store of every run (ADR-0009).
+        return rebuild_stage(self.ledger.events, self.run_id)
 
     # ── lifecycle ─────────────────────────────────────────────────────────────
 
-    def reset(self, seed: str) -> None:
-        self.ledger.reset()
+    def _cast_map(self) -> dict[str, dict[str, str | None]]:
+        """Snapshot of each agent's model binding, keyed by agent name.
+
+        Recorded on ``run.started`` so a run is self-describing — the trace alone
+        says which models played which parts (handy for sponsor-track receipts).
+        Agents without a manifest (Phase-0/1 fallback) are reported as unbound.
+        """
+        cast: dict[str, dict[str, str | None]] = {}
+        for agent in self.scenario.agents:
+            name = getattr(agent, "name", agent.__class__.__name__)
+            manifest = getattr(agent, "manifest", None)
+            cast[name] = {
+                "model_endpoint": getattr(manifest, "model_endpoint", None),
+                "model_profile": getattr(manifest, "model_profile", None),
+            }
+        return cast
+
+    def reset(self, seed: str, *, session_id: str | None = None) -> None:
+        # NOTE: we no longer wipe the ledger — it is a shared, persistent, append-only
+        # store (ADR-0009), so a reset mints a *new* run rather than destroying prior
+        # ones.  Only the in-conductor transient state for the old run is cleared.
+        #
+        # ``session_id`` (optional) attributes the run to the browser/user that started
+        # it — stamped onto ``run.started`` so the per-user Archive can list "my runs"
+        # without a side table (ADR-0014: every view is a projection of the log).
         if self.observer:
             self.observer.reset()
         self._trigger_queue.clear()
         self._pending.clear()
         self.agent_errors.clear()
         self.run_id = str(uuid4())
+        # Normalize at the engine boundary: the id originates client-side
+        # (localStorage), so malformed/oversized values degrade to None here
+        # rather than reaching the ledger or the memory index.
+        self.session_id = normalize_session_id(session_id)
         self.turn = 0
         self.governor.reset()
+        goal = getattr(self.scenario, "goal", "")
+        scenario_name = getattr(self.scenario, "name", type(self.scenario).__name__)
+        cast = self._cast_map()
         obs.set_context(run_id=self.run_id, turn=self.turn)
-        obs.log("run.started", run_id=self.run_id, seed=seed, goal=getattr(self.scenario, "goal", ""))
+        obs.log("run.started", run_id=self.run_id, seed=seed, goal=goal, scenario=scenario_name)
+        payload: dict = {"seed": seed, "goal": goal, "scenario": scenario_name, "cast": cast}
+        if self.session_id:
+            payload["session_id"] = self.session_id
         genesis_start = Event(
             run_id=self.run_id,
             turn=self.turn,
             kind="run.started",
             actor="conductor",
-            payload={"seed": seed, "goal": getattr(self.scenario, "goal", "")},
+            payload=payload,
         )
         self._append(genesis_start)
         for event in self.scenario.genesis(self.run_id, self.turn, seed):
             self._append(event)
 
+    def finalize(
+        self,
+        reason: str,
+        *,
+        winner: str | None = None,
+        winning_model: 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.
+        """
+        existing = [e for e in self.ledger.events_for_run(self.run_id) if e.kind == "run.finished"]
+        if existing:
+            return existing[0]
+        stats = self.governor.stats
+        finished = Event(
+            run_id=self.run_id,
+            turn=self.turn,
+            kind="run.finished",
+            actor="conductor",
+            payload={
+                "reason": reason,
+                "winner": winner,
+                "winning_model": winning_model,
+                "turns": int(stats.get("current_turn", self.turn) or self.turn),
+                "tokens": int(stats.get("total_tokens", 0) or 0),
+            },
+        )
+        obs.log(
+            "run.finished",
+            run_id=self.run_id,
+            reason=reason,
+            winner=winner,
+            winning_model=winning_model,
+            turns=finished.payload["turns"],
+            tokens=finished.payload["tokens"],
+        )
+        return self._append(finished)
+
     def restore(self) -> bool:
         """Resume a persisted run: adopt the ledger's run_id and last turn.
 
@@ -134,10 +213,16 @@ class Conductor:
         With an empty ledger, the first tick performs genesis instead of acting
         (preserving the original auto-reset behaviour)."""
         for _ in range(max(1, n_ticks)):
-            if not self.ledger.events:
+            if not self.ledger.events_for_run(self.run_id):
                 self.reset(self.scenario.default_seed)
                 continue
-            self._tick()
+            try:
+                self._tick()
+            except BudgetExceeded:
+                # Close the run on the ledger before the stop propagates — a headless
+                # run that hits a budget bound should still be self-describing.
+                self.finalize("budget")
+                raise
             self._maybe_snapshot()
 
     def step_one(self) -> bool:
@@ -154,23 +239,27 @@ class Conductor:
 
         Returns True when it produced an event (or performed genesis), False when the
         opened turn had no actors.  May raise :class:`BudgetExceeded` like ``step``."""
-        if not self.ledger.events:
+        if not self.ledger.events_for_run(self.run_id):
             self.reset(self.scenario.default_seed)
             return True
 
-        if not self._pending:
-            self.turn += 1
-            self.governor.begin_turn(self.turn)
-            self.governor.check(self.turn)
-            obs.set_context(turn=self.turn)
-            self._pending.extend(agent for agent, _ in self._trigger_queue)
-            self._trigger_queue.clear()
-            self._pending.extend(self._tick_scheduled_agents())
+        try:
             if not self._pending:
-                return False
-
-        agent = self._pending.popleft()
-        self._run_agent(agent, self.projection)
+                self.turn += 1
+                self.governor.begin_turn(self.turn)
+                self.governor.check(self.turn)
+                obs.set_context(turn=self.turn)
+                self._pending.extend(agent for agent, _ in self._trigger_queue)
+                self._trigger_queue.clear()
+                self._pending.extend(self._tick_scheduled_agents())
+                if not self._pending:
+                    return False
+
+            agent = self._pending.popleft()
+            self._run_agent(agent, self.projection)
+        except BudgetExceeded:
+            self.finalize("budget")
+            raise
         # Absorb subscribers this agent's event just triggered into the current turn,
         # so a subscription cascade still resolves within the turn (as in ``_tick``).
         while self._trigger_queue:
@@ -224,7 +313,10 @@ class Conductor:
                     run_id=self.run_id,
                     turn=self.turn,
                     projection=projection,
-                    recent_events=self.ledger.events,
+                    # Run-scoped: the ledger holds EVERY run (shared store, ADR-0026);
+                    # an agent's memory/context must never recall another run's — or
+                    # another user's — discussion.
+                    recent_events=self.ledger.events_for_run(self.run_id),
                 )
             except BudgetExceeded:
                 raise  # an intentional stop from the governor — never swallow it
@@ -261,6 +353,11 @@ class Conductor:
             snapshot_to(self.snapshot_path)
 
     def _append(self, event: Event) -> Event:
+        # Stamp the session onto the envelope at the single append chokepoint, so
+        # *every* action in a run is attributable/filterable by who drove it —
+        # agents and scenarios never have to know sessions exist.
+        if self.session_id and event.session_id is None:
+            event = event.model_copy(update={"session_id": self.session_id})
         appended = self.ledger.append(event)
         obs.log(
             "event.append", level="debug", id=appended.id, kind=appended.kind, actor=appended.actor, turn=appended.turn

src/core/events.py

@@ -20,9 +20,26 @@ from src import observability as obs
 # A kind is a lowercase, dot-namespaced identifier with at least two segments:
 #     <namespace>.<name>   e.g. "agent.spoke", "clue.found", "episode.published"
 
+# Documented core payload shapes (the schema validates kind *format*, not payload —
+# these are conventions for the engine's own events):
+#
+#   run.finished payload:
+#       {
+#           "reason": "verdict" | "budget" | "tick_cap" | "user_stop",
+#           "winner": str | None,           # actor name of the winner, if any
+#           "winning_model": str | None,    # model bound to the winner, if known
+#           "turns": int,                   # turns elapsed in the run
+#           "tokens": int,                  # total tokens spent in the run
+#       }
+#
+#   run.started payload is being ENRICHED in a later step (scenario name +
+#   cast->model map).  That change is purely ADDITIVE — new keys alongside the
+#   existing ones — so `schema_version` stays 1 (no migration required).
+
 CORE_EVENT_KINDS: frozenset[str] = frozenset(
     {
         "run.started",
+        "run.finished",
         "world.observed",
         "agent.thought",
         "agent.spoke",
@@ -44,6 +61,33 @@ def is_valid_kind(kind: str) -> bool:
     return bool(_KIND_PATTERN.match(kind))
 
 
+# ── session ids ───────────────────────────────────────────────────────────────
+#
+# A session id attributes a run (and, via the envelope below, every event in it)
+# to the browser/user that started it.  The value originates client-side
+# (localStorage), so it is UNTRUSTED input: normalize at the engine boundary
+# before it ever reaches the ledger or the memory index.
+
+_SESSION_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$")
+
+
+def normalize_session_id(value: str | None) -> str | None:
+    """Return a safe session id, or ``None`` when *value* is absent or malformed.
+
+    Accepts the ids we mint (UUIDs, ``sess-…``) and rejects anything else —
+    over-long strings, whitespace, control characters, separators that could
+    confuse downstream filters.  Rejection degrades to an unattributed run
+    rather than an error, so a tampered localStorage never breaks Summon.
+    """
+    candidate = (value or "").strip()
+    if not candidate:
+        return None
+    if _SESSION_ID_PATTERN.match(candidate):
+        return candidate
+    obs.log("session.id_rejected", level="warning", length=len(candidate))
+    return None
+
+
 class Event(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -55,6 +99,11 @@ class Event(BaseModel):
     payload: dict[str, Any]
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
     schema_version: int = 1
+    # Who (which browser/user session) drove the run this event belongs to.
+    # Stamped by the Conductor on append (single chokepoint), nullable for
+    # headless/legacy events.  An OPTIONAL envelope field — additive, so
+    # schema_version stays 1 and old rows load with session_id=None.
+    session_id: str | None = None
 
     @field_validator("kind")
     @classmethod

src/core/ledger.py

@@ -34,3 +34,14 @@ class Ledger:
         obs.log("ledger.reset", level="debug", events=len(self._events))
         self._events.clear()
         self._seen_ids.clear()
+
+    def events_for_run(self, run_id: str) -> tuple[Event, ...]:
+        """Return the events of *run_id* in append/offset order."""
+        return tuple(e for e in self._events if e.run_id == run_id)
+
+    def runs(self) -> tuple[str, ...]:
+        """Return the distinct run_ids in first-seen order."""
+        seen: dict[str, None] = {}
+        for e in self._events:
+            seen.setdefault(e.run_id, None)
+        return tuple(seen)

src/core/memory.py

@@ -88,7 +88,9 @@ class EpisodicMemory:
     agent_name: str
     max_recent: int = 8
 
-    def visible(self, events: tuple[Event, ...]) -> list[Event]:
+    def visible(self, events: tuple[Event, ...], run_id: str | None = None) -> list[Event]:
+        if run_id is not None:
+            events = tuple(e for e in events if e.run_id == run_id)
         result = [e for e in events if e.actor == self.agent_name or e.kind in _GLOBALLY_VISIBLE]
         return result[-self.max_recent :]
 
@@ -181,12 +183,19 @@ class SalienceMemory:
         """
         if self.index is None or not query or not candidates:
             return None
+        # Scope the semantic search to the candidates' run: the index spans every
+        # run in the shared store, and unscoped hits from other runs (other users'
+        # shows) would crowd the recall budget out of this run's events.  Derived
+        # from the candidates so callers that already pass a single-run slice (the
+        # conductor does) get scoping for free.
+        run_ids = {e.run_id for e in candidates}
+        run_id = next(iter(run_ids)) if len(run_ids) == 1 else None
         # The index is a derived, rebuildable lens (ADR-0018) — never load-bearing.
         # If it hiccups (a flaky hosted backend, a transient mem0 error), degrade to
         # keyword relevance rather than let one agent's recall crash its whole turn.
         try:
             self.index.index(tuple(candidates))
-            hits = self.index.search(query, k=len(candidates))
+            hits = self.index.search(query, k=len(candidates), run_id=run_id)
         except Exception as exc:  # noqa: BLE001 — relevance is best-effort, never fatal
             logger.warning("memory index unavailable, using keyword relevance: %s", exc)
             obs.log("memory.index.fallback", level="warning", agent=self.agent_name, error=str(exc))
@@ -198,7 +207,11 @@ class SalienceMemory:
         n = len(ranked)
         return {eid: (n - i) / n for i, eid in enumerate(ranked)}
 
-    def visible(self, events: tuple[Event, ...], current_turn: int, query: str) -> list[Event]:
+    def visible(
+        self, events: tuple[Event, ...], current_turn: int, query: str, run_id: str | None = None
+    ) -> list[Event]:
+        if run_id is not None:
+            events = tuple(e for e in events if e.run_id == run_id)
         candidates = self._candidates(events)
         relevance = self._relevance_map(candidates, query)
         scored = sorted(

src/core/memory_index.py

@@ -91,8 +91,13 @@ class MemoryIndex(Protocol):
         """Derive/refresh index entries for *events* (idempotent by ``event.id``)."""
         ...
 
-    def search(self, query: str, k: int) -> list[Event]:
-        """Return up to *k* indexed events most semantically relevant to *query*."""
+    def search(self, query: str, k: int, run_id: str | None = None) -> list[Event]:
+        """Return up to *k* indexed events most semantically relevant to *query*.
+
+        When *run_id* is given, hits MUST be scoped to that run — the ledger holds
+        every run, and recall across runs would leak one show's (or one user's)
+        discussion into another's context.
+        """
         ...
 
 
@@ -134,7 +139,7 @@ class _Mem0BackendBase:
         """Upsert one event verbatim into *mem* (``infer=False``; ledger is truth)."""
         raise NotImplementedError
 
-    def _query(self, mem: object, query: str, k: int) -> list[dict]:
+    def _query(self, mem: object, query: str, k: int, run_id: str | None) -> list[dict]:
         """Run semantic search on *mem*; return raw hit dicts (carrying metadata)."""
         raise NotImplementedError
 
@@ -160,8 +165,13 @@ class _Mem0BackendBase:
             self._store(mem, event)
             self._indexed.add(event.id)
 
-    def search(self, query: str, k: int) -> list[Event]:
-        """Semantic search; map hits back to :class:`Event` via stored metadata."""
+    def search(self, query: str, k: int, run_id: str | None = None) -> list[Event]:
+        """Semantic search; map hits back to :class:`Event` via stored metadata.
+
+        *run_id* scopes recall to one run, filtered both natively (mem0's own
+        ``run_id`` identity, pushed down to the vector store) and defensively
+        here on the reconstructed event — belt and suspenders against backends
+        that ignore unknown filters."""
         if not query or k <= 0:
             return []
         with obs.span(
@@ -171,10 +181,13 @@ class _Mem0BackendBase:
             started = time.perf_counter()
             mem = self._memory()
             events: list[Event] = []
-            for hit in self._query(mem, query, k):
+            for hit in self._query(mem, query, k, run_id):
                 event = _event_from_metadata(hit.get("metadata"))
-                if event is not None:
-                    events.append(event)
+                if event is None:
+                    continue
+                if run_id is not None and event.run_id != run_id:
+                    continue
+                events.append(event)
             elapsed_ms = (time.perf_counter() - started) * 1000
             obs.add_span_attrs(**{"memory.hits": len(events), "memory.latency_ms": round(elapsed_ms, 2)})
             obs.observe("memory.index.hits", len(events))
@@ -228,12 +241,20 @@ class Mem0MemoryIndex(_Mem0BackendBase):
         mem.add(  # type: ignore[attr-defined]
             _event_text(event),
             user_id=self._NAMESPACE,
+            # mem0's native identity scopes: run_id partitions recall per run and
+            # agent_id per actor, so filtering happens in the vector store rather
+            # than post-hoc in Python (mem0 best practice).
+            run_id=event.run_id,
+            agent_id=event.actor or None,
             metadata=_event_metadata(event),
             infer=False,  # store verbatim; the ledger, not a model, is truth
         )
 
-    def _query(self, mem: object, query: str, k: int) -> list[dict]:
-        return _result_items(mem.search(query, top_k=k, filters={"user_id": self._NAMESPACE}))  # type: ignore[attr-defined]
+    def _query(self, mem: object, query: str, k: int, run_id: str | None) -> list[dict]:
+        filters: dict = {"user_id": self._NAMESPACE}
+        if run_id:
+            filters["run_id"] = run_id
+        return _result_items(mem.search(query, top_k=k, filters=filters))  # type: ignore[attr-defined]
 
 
 # ── hosted (opt-in) backend ────────────────────────────────────────────────────
@@ -302,12 +323,17 @@ class Mem0CloudIndex(_Mem0BackendBase):
         mem.add(  # type: ignore[attr-defined]
             [{"role": "user", "content": _event_text(event)}],
             user_id=self._NAMESPACE,
+            run_id=event.run_id,  # native per-run scope (see local backend)
+            agent_id=event.actor or None,
             metadata=_event_metadata(event),
             infer=False,
         )
 
-    def _query(self, mem: object, query: str, k: int) -> list[dict]:
-        return _result_items(mem.search(query, top_k=k, filters={"user_id": self._NAMESPACE}))  # type: ignore[attr-defined]
+    def _query(self, mem: object, query: str, k: int, run_id: str | None) -> list[dict]:
+        filters: dict = {"user_id": self._NAMESPACE}
+        if run_id:
+            filters["run_id"] = run_id
+        return _result_items(mem.search(query, top_k=k, filters=filters))  # type: ignore[attr-defined]
 
 
 # ── metadata round-trip (event ⇄ vector entry) ────────────────────────────────
@@ -324,6 +350,7 @@ def _event_metadata(event: Event) -> dict:
         "payload": event.payload,
         "created_at": event.created_at.isoformat(),
         "schema_version": event.schema_version,
+        "session_id": event.session_id,
     }
 
 
@@ -340,6 +367,7 @@ def _event_from_metadata(metadata: dict | None) -> Event | None:
             actor=str(metadata.get("actor", "")),
             payload=dict(metadata.get("payload") or {}),
             schema_version=int(metadata.get("schema_version", 1)),
+            session_id=metadata.get("session_id") or None,
         )
     except (KeyError, ValueError, TypeError):  # pragma: no cover - defensive
         return None

src/core/projections.py

@@ -43,8 +43,10 @@ class StageProjection:
             self.agent_notes = self.agent_notes[-8:]
 
 
-def rebuild_stage(events: tuple[Event, ...]) -> StageProjection:
+def rebuild_stage(events: tuple[Event, ...], run_id: str | None = None) -> StageProjection:
     projection = StageProjection()
+    if run_id is not None:
+        events = tuple(e for e in events if e.run_id == run_id)
     for event in events:
         projection.apply(event)
     return projection

src/core/run_index.py

@@ -0,0 +1,135 @@
+"""Run index — a per-run summary projection over the append-only ledger.
+
+This is a *projection* in the same spirit as :func:`src.core.projections.rebuild_stage`:
+a pure fold over events that yields one :class:`RunSummary` per ``run_id``.  It reads the
+two engine bookend events — ``run.started`` (scenario / seed / cast / start time) and
+``run.finished`` (reason / winner / token+turn totals / finish time) — and ignores the
+rest.  A run that has started but not finished yields a summary with the ``finished_*``
+fields left ``None`` / zero.
+
+Two implementations, one contract:
+
+  * :func:`index_runs` is the reference oracle — a pure function over any
+    ``Iterable[Event]``, folding per ``run_id`` in first-seen order.  Tests and offline
+    callers use this.
+  * :func:`index_runs_from_ledger` produces the *same* list straight from a ledger's
+    indexed queries (``runs()`` + ``events_for_run()``), so a hosted SQL backend never
+    has to stream every event through Python to list its runs.
+
+The cast payload mirrors the enriched ``run.started`` shape (see ``src/core/events.py``):
+``{agent_name: {"model_endpoint": str | None, "model_profile": str | None}}``.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, Iterable
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from src.core.events import Event
+
+if TYPE_CHECKING:  # avoid a hard import cycle / SQLAlchemy dependency on the offline path
+    from src.core.ledger import Ledger
+
+
+class CastBinding(BaseModel):
+    """The model a cast member is bound to (mirrors the ``run.started`` cast entry)."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    model_endpoint: str | None = None
+    model_profile: str | None = None
+
+
+class RunSummary(BaseModel):
+    """A one-line summary of a single run, folded from its bookend events."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    run_id: str
+    scenario: str = ""
+    seed: str = ""
+    session_id: str | None = None
+    cast: dict[str, CastBinding] = Field(default_factory=dict)
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    reason: str | None = None
+    winner: str | None = None
+    winning_model: str | None = None
+    turns: int = 0
+    tokens: int = 0
+
+
+def _coerce_cast(raw: Any) -> dict[str, CastBinding]:
+    if not isinstance(raw, dict):
+        return {}
+    cast: dict[str, CastBinding] = {}
+    for name, binding in raw.items():
+        if isinstance(binding, dict):
+            cast[name] = CastBinding(**binding)
+        else:
+            cast[name] = CastBinding()
+    return cast
+
+
+def _apply_started(summary: RunSummary, event: Event) -> None:
+    payload = event.payload
+    summary.scenario = payload.get("scenario", "") or ""
+    summary.seed = payload.get("seed", "") or ""
+    # Attribution lives on the envelope (stamped by the conductor on every event);
+    # the run.started payload copy is kept for human-readable traces.
+    summary.session_id = event.session_id or payload.get("session_id") or None
+    summary.cast = _coerce_cast(payload.get("cast"))
+    summary.started_at = event.created_at
+
+
+def _apply_finished(summary: RunSummary, event: Event) -> None:
+    payload = event.payload
+    summary.reason = payload.get("reason")
+    summary.winner = payload.get("winner")
+    summary.winning_model = payload.get("winning_model")
+    summary.turns = int(payload.get("turns", 0) or 0)
+    summary.tokens = int(payload.get("tokens", 0) or 0)
+    summary.finished_at = event.created_at
+
+
+def index_runs(events: Iterable[Event]) -> list[RunSummary]:
+    """Fold *events* into one :class:`RunSummary` per run, in first-seen order.
+
+    Pure reference oracle: ``run.started`` populates scenario/seed/cast/started_at and
+    ``run.finished`` populates reason/winner/winning_model/turns/tokens/finished_at.
+    Runs appear in the order their *first* event is seen.  All other event kinds are
+    ignored (they don't change the summary), but any event is enough to register a run.
+    """
+    summaries: dict[str, RunSummary] = {}
+    for event in events:
+        summary = summaries.get(event.run_id)
+        if summary is None:
+            summary = RunSummary(run_id=event.run_id)
+            summaries[event.run_id] = summary
+        if event.kind == "run.started":
+            _apply_started(summary, event)
+        elif event.kind == "run.finished":
+            _apply_finished(summary, event)
+    return list(summaries.values())
+
+
+def index_runs_from_ledger(ledger: "Ledger") -> list[RunSummary]:
+    """Build the same :class:`RunSummary` list using a ledger's indexed queries.
+
+    Equivalent to ``index_runs(ledger.events)`` but reads run-scoped slices via
+    ``ledger.runs()`` + ``ledger.events_for_run()`` so SQL backends keep the work in the
+    database's indexes rather than streaming the whole log through Python.  ``index_runs``
+    remains the oracle the two paths are checked against.
+    """
+    summaries: list[RunSummary] = []
+    for run_id in ledger.runs():
+        summary = RunSummary(run_id=run_id)
+        for event in ledger.events_for_run(run_id):
+            if event.kind == "run.started":
+                _apply_started(summary, event)
+            elif event.kind == "run.finished":
+                _apply_finished(summary, event)
+        summaries.append(summary)
+    return summaries

src/core/sqlalchemy_ledger.py

@@ -26,6 +26,7 @@ installed.  The offline in-memory path stays import-clean.
 Backend selection lives in :mod:`src.core.ledger_factory`: with ``DATABASE_URL``
 set the system uses this store; otherwise it uses the in-memory ``Ledger``.
 """
+
 from __future__ import annotations
 
 import json
@@ -95,6 +96,7 @@ class SqlAlchemyLedger(Ledger):
             Column("payload", Text, nullable=False),
             Column("created_at", DateTime(timezone=True), nullable=False),
             Column("schema_version", Integer, nullable=False, server_default="1"),
+            Column("session_id", String, nullable=True, index=True),
         )
         self._metadata.create_all(self._engine)
 
@@ -121,6 +123,7 @@ class SqlAlchemyLedger(Ledger):
                         payload=json.dumps(event.payload),
                         created_at=_aware(event.created_at),
                         schema_version=event.schema_version,
+                        session_id=event.session_id,
                     )
                 )
             self._cache.append(event)
@@ -179,6 +182,26 @@ class SqlAlchemyLedger(Ledger):
             value = conn.execute(stmt).scalar()
         return value or 0
 
+    def events_for_run(self, run_id: str) -> tuple[Event, ...]:
+        """Return the events of *run_id* in append/offset order (indexed query)."""
+        from sqlalchemy import select
+
+        t = self._events_table
+        stmt = select(t).where(t.c.run_id == run_id).order_by(t.c.offset)
+        with self._engine.connect() as conn:
+            rows = conn.execute(stmt).mappings().all()
+        return tuple(self._row_to_event(row) for row in rows)
+
+    def runs(self) -> tuple[str, ...]:
+        """Return the distinct run_ids in first-seen order (indexed query)."""
+        from sqlalchemy import func, select
+
+        t = self._events_table
+        stmt = select(t.c.run_id).group_by(t.c.run_id).order_by(func.min(t.c.offset))
+        with self._engine.connect() as conn:
+            rows = conn.execute(stmt).scalars().all()
+        return tuple(rows)
+
     def close(self) -> None:
         self._engine.dispose()
 
@@ -219,6 +242,7 @@ class SqlAlchemyLedger(Ledger):
             payload=payload,
             created_at=created_at,
             schema_version=row["schema_version"],
+            session_id=row.get("session_id"),
         )
 
 

src/core/sqlite_ledger.py

@@ -13,6 +13,7 @@ Design decisions:
   - reset() is deliberately destructive: it clears the current run, not all runs.
     Multi-run persistence (keeping history across resets) is a Phase 3 milestone.
 """
+
 from __future__ import annotations
 
 import json
@@ -55,11 +56,13 @@ class SQLiteLedger(Ledger):
                 actor           TEXT    NOT NULL,
                 payload         TEXT    NOT NULL,
                 created_at      TEXT    NOT NULL,
-                schema_version  INTEGER NOT NULL DEFAULT 1
+                schema_version  INTEGER NOT NULL DEFAULT 1,
+                session_id      TEXT
             );
             CREATE INDEX IF NOT EXISTS idx_run_id ON events(run_id);
             CREATE INDEX IF NOT EXISTS idx_kind   ON events(kind);
             CREATE INDEX IF NOT EXISTS idx_actor  ON events(actor);
+            CREATE INDEX IF NOT EXISTS idx_session_id ON events(session_id);
         """)
         self._conn.commit()
 
@@ -71,8 +74,8 @@ class SQLiteLedger(Ledger):
         try:
             self._conn.execute(
                 "INSERT INTO events "
-                "(id, run_id, turn, kind, actor, payload, created_at, schema_version) "
-                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                "(id, run_id, turn, kind, actor, payload, created_at, schema_version, session_id) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                 (
                     event.id,
                     event.run_id,
@@ -82,6 +85,7 @@ class SQLiteLedger(Ledger):
                     json.dumps(event.payload),
                     event.created_at.isoformat(),
                     event.schema_version,
+                    event.session_id,
                 ),
             )
             self._conn.commit()
@@ -118,52 +122,59 @@ class SQLiteLedger(Ledger):
         ledger = cls(path)
         return ledger
 
+    _SELECT_COLS = "id, run_id, turn, kind, actor, payload, created_at, schema_version, session_id"
+
+    @staticmethod
+    def _row_to_event(row: tuple) -> Event:
+        try:
+            created_at = datetime.fromisoformat(row[6])
+            if created_at.tzinfo is None:
+                created_at = created_at.replace(tzinfo=timezone.utc)
+        except (ValueError, TypeError):
+            created_at = datetime.now(timezone.utc)
+        return Event(
+            id=row[0],
+            run_id=row[1],
+            turn=row[2],
+            kind=row[3],  # type: ignore[arg-type]
+            actor=row[4],
+            payload=json.loads(row[5]),
+            created_at=created_at,
+            schema_version=row[7],
+            session_id=row[8],
+        )
+
     def _load_cache(self) -> None:
-        rows = self._conn.execute(
-            "SELECT id, run_id, turn, kind, actor, payload, created_at, schema_version "
-            "FROM events ORDER BY offset"
-        ).fetchall()
+        rows = self._conn.execute(f"SELECT {self._SELECT_COLS} FROM events ORDER BY offset").fetchall()
         for row in rows:
-            try:
-                created_at = datetime.fromisoformat(row[6])
-                if created_at.tzinfo is None:
-                    created_at = created_at.replace(tzinfo=timezone.utc)
-            except (ValueError, TypeError):
-                created_at = datetime.now(timezone.utc)
-
-            e = Event(
-                id=row[0],
-                run_id=row[1],
-                turn=row[2],
-                kind=row[3],  # type: ignore[arg-type]
-                actor=row[4],
-                payload=json.loads(row[5]),
-                created_at=created_at,
-                schema_version=row[7],
-            )
+            e = self._row_to_event(row)
             self._cache.append(e)
             self._seen_ids.add(e.id)
 
     def tail(self, from_offset: int = 0) -> tuple[Event, ...]:
         """Return events with offset > from_offset (for crash-recovery replay)."""
         rows = self._conn.execute(
-            "SELECT id, run_id, turn, kind, actor, payload, created_at, schema_version "
-            "FROM events WHERE offset > ? ORDER BY offset",
+            f"SELECT {self._SELECT_COLS} FROM events WHERE offset > ? ORDER BY offset",
             (from_offset,),
         ).fetchall()
-        events = []
-        for row in rows:
-            e = Event(
-                id=row[0], run_id=row[1], turn=row[2], kind=row[3],  # type: ignore[arg-type]
-                actor=row[4], payload=json.loads(row[5]),
-                schema_version=row[7],
-            )
-            events.append(e)
-        return tuple(events)
+        return tuple(self._row_to_event(row) for row in rows)
 
     def latest_offset(self) -> int:
         row = self._conn.execute("SELECT MAX(offset) FROM events").fetchone()
         return row[0] or 0
 
+    def events_for_run(self, run_id: str) -> tuple[Event, ...]:
+        """Return the events of *run_id* in append/offset order (indexed query)."""
+        rows = self._conn.execute(
+            f"SELECT {self._SELECT_COLS} FROM events WHERE run_id = ? ORDER BY offset",
+            (run_id,),
+        ).fetchall()
+        return tuple(self._row_to_event(row) for row in rows)
+
+    def runs(self) -> tuple[str, ...]:
+        """Return the distinct run_ids in first-seen order (indexed query)."""
+        rows = self._conn.execute("SELECT run_id FROM events GROUP BY run_id ORDER BY MIN(offset)").fetchall()
+        return tuple(row[0] for row in rows)
+
     def close(self) -> None:
         self._conn.close()

src/ui/fishbowl/app.py

@@ -30,6 +30,20 @@ from src.tools.builtins import default_tool_registry
 from src.ui.fishbowl.adapter import scenario_voice
 from src.ui.fishbowl.view_model import view_model_at
 
+try:  # archive: "my past sessions" read layer over the run-history ledger (ADR-0026)
+    from src.ui.fishbowl.archive import list_runs, load_replay, run_card_label
+except Exception:  # pragma: no cover - degrade gracefully if the archive unit is absent
+
+    def list_runs(*_args, **_kwargs):
+        return []
+
+    def load_replay(*_args, **_kwargs):
+        return None
+
+    def run_card_label(_summary):
+        return "▶ (run)"
+
+
 # ── loop-safety backstop ────────────────────────────────────────────────────────
 # Belt-and-suspenders against a runaway autoplay loop: even when the governor never
 # trips (e.g. a generous budget) the timer halts after this many consecutive auto-ticks
@@ -384,6 +398,12 @@ def advance_one_tick(session: FishbowlSession | None, k: int, ticks: int, *, max
     ticks = int(ticks or 0)
     if session is None:
         return 0, 0, None
+    if getattr(session, "replay", False):
+        # A loaded past run: replay forward through the recorded prefix, then stop.
+        # It owns no live engine, so generating is impossible — never spend a tick here.
+        if k < session.head:
+            return k + 1, ticks, None
+        return k, ticks, "end of session — replay complete"
     if session.has_verdict():
         return k, ticks, "verdict reached — the show resolved"
     if k < session.head:
@@ -479,10 +499,33 @@ def build_app() -> gr.Blocks:
         blank_state = gr.State("")  # stand-in input when a leaf widget is absent
         stopped_state = gr.State(False)  # set once the run halts (budget/backstop/verdict)
         tick_count_state = gr.State(0)  # consecutive autoplay ticks (the 40-tick backstop)
+        # Per-user session id: resolved from the browser's localStorage on load (see
+        # _SESSION_ID_JS) so it survives reloads.  Stamped onto run.started so the
+        # Archive can list "my sessions only"; a hidden carrier, never shown.
+        session_id_box = gr.Textbox(value="", visible=False, elem_id="fb-session-id")
+
+        # Populated after the tabs build, then read at gr.render runtime (post-load) so
+        # the Lab's Archive drawer can target the Show's panes it lists into.
+        archive_refs: dict = {}
 
         with gr.Tabs() as tabs:
             with gr.Tab("The Lab", id="lab"):
                 lab_handles = build_lab()
+                _build_archive_drawer(
+                    scenario_handle=(lab_handles or {}).get("scenario"),
+                    session_id_box=session_id_box,
+                    refs=archive_refs,
+                    tabs=tabs,
+                    states={
+                        "session": session_state,
+                        "k": k_state,
+                        "scenario": scenario_state,
+                        "layout": layout_state,
+                        "mind": mind_reader_state,
+                        "stopped": stopped_state,
+                        "ticks": tick_count_state,
+                    },
+                )
             with gr.Tab("The Show", id="show"):
                 show_handles = build_show()
             with gr.Tab("Telemetry", id="telemetry"):
@@ -491,6 +534,10 @@ def build_app() -> gr.Blocks:
         # CRT foreground layers (scanlines + vignette, above content, click-through).
         gr.HTML(_CRT_FG_HTML)
 
+        # The Archive's gr.render runs client-side after build; by then show_handles
+        # exists, so it reads the live panes through this ref.
+        archive_refs["show_handles"] = show_handles or {}
+
         _wire(
             tabs=tabs,
             lab_handles=lab_handles or {},
@@ -503,8 +550,13 @@ def build_app() -> gr.Blocks:
             blank_state=blank_state,
             stopped_state=stopped_state,
             tick_count_state=tick_count_state,
+            session_id_box=session_id_box,
         )
 
+        # Resolve (or mint) the browser's session id once the page loads; updating the
+        # hidden box fires its change, which re-renders the Archive list for this user.
+        demo.load(None, None, [session_id_box], js=_SESSION_ID_JS)
+
     return demo
 
 
@@ -532,6 +584,7 @@ def _wire(
     blank_state: gr.State,
     stopped_state: gr.State,
     tick_count_state: gr.State,
+    session_id_box: gr.Textbox | None = None,
 ) -> None:
     """Connect Lab/Show component handles to session transport + HTML re-render.
 
@@ -688,6 +741,7 @@ def _wire(
         hourly_budget_usd,
         layout,
         mind_reader,
+        session_id,
     ):
         title = _scenario_title(scenario_value)
         name = SCENARIOS.get(title, "")
@@ -716,7 +770,7 @@ def _wire(
             else None
         )
         if session is not None:
-            session.reset((seed_value or "").strip())
+            session.reset((seed_value or "").strip(), session_id=(session_id or "").strip() or None)
             k = session.head
         else:
             k = 0
@@ -746,6 +800,7 @@ def _wire(
             hourly_budget_in if hourly_budget_in is not None else blank_state,
             layout_state,
             mind_reader_state,
+            session_id_box if session_id_box is not None else blank_state,
         ]
         summon_btn.click(
             on_summon,
@@ -962,6 +1017,123 @@ def _pad_values(values, show_outs: list) -> tuple:
     return tuple(values[: len(show_outs)])
 
 
+# ── Archive drawer ("my past sessions" → read-only replay) ──────────────────────
+
+# Resolve (or mint) a per-browser session id from localStorage on page load.  Kept in
+# JS so the id is the user's own and survives reloads — Python only reads the carrier.
+_SESSION_ID_JS = """
+() => {
+  try {
+    let id = localStorage.getItem('fishbowl_session_id');
+    if (!id) {
+      id = (window.crypto && crypto.randomUUID)
+        ? crypto.randomUUID()
+        : 'sess-' + Math.random().toString(36).slice(2) + Date.now().toString(36);
+      localStorage.setItem('fishbowl_session_id', id);
+    }
+    return id;
+  } catch (e) {
+    return 'sess-' + Math.random().toString(36).slice(2);
+  }
+}
+"""
+
+
+def _title_for(value) -> str:
+    """Resolve a scenario title *or* internal name to a display-title key."""
+    if value in SCENARIOS:
+        return value
+    for title, name in SCENARIOS.items():
+        if value == name:
+            return title
+    return _DEFAULT_TITLE
+
+
+def _show_outs(show_handles: dict) -> list:
+    """The present Show panes (stage, feed, meters, verdict), in render order."""
+    stage = _h(show_handles, "stage", "stage_html", "constellation")
+    feed = _h(show_handles, "feed", "feed_html")
+    meters = _h(show_handles, "meters", "meters_html")
+    verdict = _h(show_handles, "verdict", "verdict_html")
+    return [c for c in (stage, feed, meters, verdict) if c is not None]
+
+
+def _archive_empty_html() -> str:
+    """The drawer's empty state — no past runs for this world yet."""
+    return _fishbowl(
+        '<div class="archive-empty">'
+        '<div class="eyebrow">&#10227; No past sessions</div>'
+        '<div class="ae-body">Summon the bowl, and your runs in this world '
+        "will gather here — yours alone, replayable any time.</div>"
+        "</div>",
+        role="fb-archive",
+    )
+
+
+def _build_archive_drawer(*, scenario_handle, session_id_box, refs: dict, tabs, states: dict) -> None:
+    """The Lab's "Past sessions" accordion: clickable phosphor cards → read-only replay.
+
+    A ``gr.render`` keyed on (scenario, session id) lists *this user's* runs for the
+    *current* world via :func:`list_runs`; clicking a card loads that run with
+    :func:`load_replay` and jumps to the Show.  The list re-renders on world change,
+    when the session id resolves from localStorage, and on the manual refresh.
+    """
+    if scenario_handle is None:  # no scenario picker → nothing to scope a list to
+        return
+
+    with gr.Accordion("⟲ Past sessions · this world", open=False, elem_classes=["archive-drawer"]):
+        refresh = gr.Button("⟳ refresh", size="sm", elem_classes=["archive-refresh"], scale=0)
+
+        @gr.render(
+            inputs=[scenario_handle, session_id_box],
+            triggers=[scenario_handle.change, session_id_box.change, refresh.click],
+        )
+        def _render_archive(scenario_value, session_id):
+            name = SCENARIOS.get(_title_for(scenario_value), "")
+            runs = list_runs(name, session_id)
+            if not runs:
+                gr.HTML(_archive_empty_html())
+                return
+
+            show_outs = _show_outs(refs.get("show_handles") or {})
+            n_out = 6 + len(show_outs)  # session, k, scenario, tabs, *panes, stopped, ticks
+
+            def _loader(run_id: str):
+                def _load(layout, mind_reader):
+                    session = load_replay(run_id, registry=_registry, tools=_tools)
+                    if session is None:
+                        return tuple(gr.update() for _ in range(n_out))
+                    k = session.head  # land on the full discussion; scrub/▶ replays it
+                    out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
+                    return (
+                        session,
+                        k,
+                        _title_for(session.scenario_name),
+                        gr.update(selected="show"),
+                        *_pad_values(out, show_outs),
+                        False,
+                        0,
+                    )
+
+                return _load
+
+            for summary in runs:
+                card = gr.Button(run_card_label(summary), elem_classes=["archive-card"])
+                card.click(
+                    _loader(summary.run_id),
+                    inputs=[states["layout"], states["mind"]],
+                    outputs=[
+                        states["session"],
+                        states["k"],
+                        states["scenario"],
+                        tabs,
+                        *show_outs,
+                        states["stopped"],
+                        states["ticks"],
+                    ],
+                )
+
+
 # ── dev server port (ported from the original root app.py) ──────────────────────
 
 

src/ui/fishbowl/archive.py

@@ -0,0 +1,145 @@
+"""The Archive — "my past sessions" for the Lab's Load drawer.
+
+This is a thin read layer over the W1 run-history surface (ADR-0026): it folds the
+shared ledger into per-run :class:`RunSummary` rows via
+:func:`src.core.run_index.index_runs_from_ledger`, then keeps only the runs that
+belong to *this* browser/user (``session_id``) in *this* world (``scenario``).  No
+side table is introduced — the run list is a projection of the log, in keeping with
+the single-source-of-truth discipline (ADR-0014).
+
+Loading a run hands back a :class:`ReplaySession`: a read-only view the Show scrubs
+without ever generating, so a past session replays verbatim with no token spend.
+"""
+
+from __future__ import annotations
+
+from src import observability as obs
+from src.core.events import normalize_session_id
+from src.core.ledger_factory import make_ledger
+from src.core.registry import Registry, default_registry
+from src.core.run_index import RunSummary, index_runs_from_ledger
+from src.ui.fishbowl.session import ReplaySession
+
+
+def _sort_key(summary: RunSummary):
+    # Newest first; runs missing a start time sink to the bottom deterministically.
+    started = summary.started_at
+    return (started is not None, started)
+
+
+def list_runs(
+    scenario_name: str,
+    session_id: str | None,
+    *,
+    registry: Registry | None = None,
+    limit: int = 50,
+) -> list[RunSummary]:
+    """Past runs for *scenario_name* started by *session_id*, newest first.
+
+    Returns ``[]`` when the world or the session is unknown — the drawer then shows
+    its empty state rather than every user's runs.  Scoping to ``session_id`` is what
+    makes "my sessions only" hold across browsers (each user has their own id).
+    """
+    session_id = normalize_session_id(session_id)  # same boundary rule as the write path
+    if not scenario_name or not session_id:
+        return []
+    try:
+        ledger = make_ledger()
+    except Exception:  # pragma: no cover - no event store configured (defensive)
+        return []
+    summaries = [
+        s for s in index_runs_from_ledger(ledger) if s.scenario == scenario_name and s.session_id == session_id
+    ]
+    summaries.sort(key=_sort_key, reverse=True)
+    return summaries[:limit]
+
+
+def load_replay(
+    run_id: str,
+    *,
+    registry: Registry | None = None,
+    tools=None,
+) -> ReplaySession | None:
+    """Build a read-only :class:`ReplaySession` for *run_id*, or ``None`` if unknown.
+
+    The run names its own scenario on ``run.started`` (ADR-0026), so a replay is
+    self-describing — we rebuild the cast/meters from that scenario and replay the
+    recorded events verbatim.
+    """
+    if not run_id:
+        return None
+    try:
+        ledger = make_ledger()
+    except Exception:  # pragma: no cover - defensive
+        return None
+    events = ledger.events_for_run(run_id)
+    if not events:
+        return None
+    started = next((e for e in events if e.kind == "run.started"), None)
+    scenario_name = (started.payload.get("scenario") if started is not None else None) or ""
+    if not scenario_name:
+        return None
+    try:
+        return ReplaySession(
+            run_id=run_id,
+            events=events,
+            scenario_name=scenario_name,
+            registry=registry or default_registry(),
+            tools=tools,
+        )
+    except Exception:  # pragma: no cover - scenario no longer in the registry, etc.
+        obs.log("archive.load_failed", level="warning", run_id=run_id, scenario=scenario_name)
+        return None
+
+
+# ── card formatting (the phosphor list rows) ──────────────────────────────────────
+
+
+def _short_id(run_id: str) -> str:
+    """A stable, glanceable handle for a run — last 4 hex of its uuid."""
+    tail = run_id.replace("-", "")[-4:] if run_id else "????"
+    return f"#{tail}"
+
+
+def _fmt_tokens(tokens: int) -> str:
+    if tokens >= 1000:
+        return f"{tokens / 1000:.1f}k tok"
+    return f"{tokens} tok"
+
+
+def _fmt_when(summary: RunSummary) -> str:
+    started = summary.started_at
+    if started is None:
+        return "—"
+    try:
+        return started.strftime("%b %-d · %H:%M")
+    except ValueError:  # pragma: no cover - platforms without %-d
+        return started.strftime("%b %d · %H:%M")
+
+
+_REASON_LABEL = {
+    "verdict": "verdict",
+    "budget": "budget",
+    "tick_cap": "tick-cap",
+    "user_stop": "stopped",
+}
+
+
+def run_card_label(summary: RunSummary) -> str:
+    """A compact, single-line label for a run's Load button (styled by CSS).
+
+    Shape: ``▶ #a3f9 · Jun 12 · 14:03 · 14t · WON Fox · verdict``.  An unfinished run
+    reads ``…live`` so the list distinguishes resolved runs from abandoned ones.
+    """
+    parts = [f"▶ {_short_id(summary.run_id)}", _fmt_when(summary)]
+    if summary.turns:
+        parts.append(f"{summary.turns}t")
+    if summary.tokens:
+        parts.append(_fmt_tokens(summary.tokens))
+    if summary.winner:
+        parts.append(f"WON {summary.winner}")
+    if summary.reason:
+        parts.append(_REASON_LABEL.get(summary.reason, summary.reason))
+    else:
+        parts.append("…live")
+    return "  ·  ".join(parts)

src/ui/fishbowl/assets/styles.css

@@ -1181,3 +1181,85 @@ footer { display: none !important; }
   color: var(--ink) !important;
   box-shadow: 0 0 14px -4px rgba(79,230,210,0.5) !important;
 }
+
+/* ── Archive drawer: "my past sessions" (Lab → Load → read-only replay) ──────────
+   The accordion + cards are real Gradio components (not inside the `.fishbowl`
+   scope root), so these rules target their `elem_classes` directly and lean on
+   `!important` to win over Gradio's own cascade, like the seed-edit rules above. */
+.gradio-container .archive-drawer {
+  border: 1px solid var(--line-soft) !important;
+  border-radius: var(--r-lg) !important;
+  background: linear-gradient(180deg, rgba(10,34,43,0.55), rgba(5,18,26,0.35)) !important;
+  margin-top: 4px !important;
+}
+.gradio-container .archive-drawer > .label-wrap,
+.gradio-container .archive-drawer button.label-wrap,
+.gradio-container .archive-drawer span.label-wrap {
+  font-family: var(--font-display) !important;
+  font-size: 11px !important;
+  letter-spacing: 0.16em !important;
+  text-transform: uppercase !important;
+  color: var(--ink-mid) !important;
+}
+
+.gradio-container .archive-refresh {
+  min-width: 0 !important;
+  align-self: flex-end !important;
+  font-family: var(--font-display) !important;
+  font-size: 9.5px !important;
+  letter-spacing: 0.18em !important;
+  text-transform: uppercase !important;
+  color: var(--ink-dim) !important;
+  background: transparent !important;
+  border: 1px solid var(--line-soft) !important;
+  border-radius: var(--r) !important;
+}
+.gradio-container .archive-refresh:hover {
+  color: var(--cyan) !important;
+  border-color: var(--line-hot) !important;
+  box-shadow: var(--glow) !important;
+}
+
+/* Each past run is a full-width, left-aligned monospace card with a phosphor
+   spine.  The whole row is the click target (it loads the replay). */
+.gradio-container .archive-card {
+  display: block !important;
+  width: 100% !important;
+  text-align: left !important;
+  font-family: var(--font-body) !important;
+  font-size: 12px !important;
+  letter-spacing: 0.02em !important;
+  color: var(--ink) !important;
+  background: var(--panel) !important;
+  border: 1px solid var(--line-soft) !important;
+  border-left: 2px solid var(--teal) !important;
+  border-radius: var(--r) !important;
+  padding: 10px 12px !important;
+  margin: 6px 0 0 !important;
+  white-space: nowrap !important;
+  overflow: hidden !important;
+  text-overflow: ellipsis !important;
+  transition: border-color 0.16s ease, box-shadow 0.16s ease, transform 0.16s ease !important;
+}
+.gradio-container .archive-card:hover {
+  border-color: var(--line-hot) !important;
+  border-left-color: var(--cyan) !important;
+  color: var(--cyan) !important;
+  box-shadow: var(--glow) !important;
+  transform: translateX(2px) !important;
+}
+
+.fishbowl .archive-empty {
+  border: 1px dashed var(--line-soft);
+  border-radius: var(--r-lg);
+  padding: 18px 16px;
+  background: rgba(5, 18, 26, 0.35);
+}
+.fishbowl .archive-empty .ae-body {
+  font-family: var(--font-body);
+  font-size: 12px;
+  color: var(--ink-dim);
+  margin-top: 6px;
+  line-height: 1.5;
+  max-width: 46ch;
+}

src/ui/fishbowl/session.py

@@ -59,9 +59,9 @@ class FishbowlSession:
 
     # ── lifecycle ───────────────────────────────────────────────────────────────
 
-    def reset(self, seed: str = "") -> None:
+    def reset(self, seed: str = "", *, session_id: str | None = None) -> None:
         obs.log("session.reset", scenario=self._scenario_name, seed=seed or self.scenario.default_seed)
-        self.conductor.reset(seed or self.scenario.default_seed)
+        self.conductor.reset(seed or self.scenario.default_seed, session_id=session_id)
 
     def step(self, n_ticks: int = 1) -> None:
         with obs.span("session.step", **{"session.n_ticks": n_ticks}):
@@ -86,21 +86,50 @@ class FishbowlSession:
     def scenario(self):
         return self.conductor.scenario
 
+    # Marks the live, generative session apart from a read-only ``ReplaySession``;
+    # the autoplay loop checks this so loading a past run never spends tokens.
+    replay = False
+
     @property
     def events(self):
-        return self.conductor.ledger.events
+        # Run-scoped: the ledger is a shared store of *every* run (ADR-0009), so the
+        # Show must only ever see the current run's events — otherwise scenario B's
+        # stage would replay scenario A's discussion.  Every read below (head,
+        # snapshot, scrubber) flows from this, so scoping here scopes the whole Show.
+        return self.conductor.ledger.events_for_run(self.conductor.run_id)
 
     @property
     def head(self) -> int:
-        """The generation-head: number of events in the ledger so far."""
-        return len(self.conductor.ledger.events)
+        """The generation-head: number of events in *this run* so far."""
+        return len(self.events)
 
     def has_verdict(self) -> bool:
-        """True once a ``judge.verdict`` event sits in the ledger — the show resolved.
-
-        The Fishbowl autoplay loop consults this to auto-pause the timer when the
-        Judge has ruled, so the curtain falls on its own (no extra token spend)."""
-        return any(getattr(e, "kind", None) == "judge.verdict" for e in self.conductor.ledger.events)
+        """True once a ``judge.verdict`` event sits in *this run* — the show resolved.
+
+        Run-scoped (ADR-0009): the ledger is a shared, append-only store of every run,
+        so we only consult the current run's events.  The Fishbowl autoplay loop calls
+        this to auto-pause the timer when the Judge has ruled, so the curtain falls on
+        its own (no extra token spend)."""
+        return any(e.kind == "judge.verdict" for e in self.conductor.ledger.events_for_run(self.conductor.run_id))
+
+    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."""
+        winner: str | None = None
+        winning_model: str | None = None
+        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)
+            if verdict is not None:
+                winner = verdict.payload.get("winner") or None
+            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)
 
     @property
     def cast(self) -> list[AgentManifest]:
@@ -152,3 +181,97 @@ class FishbowlSession:
             token_ceiling=self.token_ceiling,
             max_rounds=self.max_rounds,
         )
+
+
+class ReplaySession:
+    """A read-only view over one *past* run — the Archive's "Load" target.
+
+    It exposes the exact read surface the Show renders (``events`` / ``head`` /
+    ``snapshot`` / ``has_verdict``) so the transport's scrubber and replay just work,
+    but it owns no live ``Conductor``: ``step``/``step_one``/``inject`` are no-ops and
+    ``replay`` is ``True``, so autoplay never generates (no token spend) on a load.
+
+    The fixed event list is the run's own slice (``events_for_run(run_id)``); the cast
+    cards / meters bounds are rebuilt from that run's scenario via the registry, while
+    the discussion itself is replayed verbatim from the events.
+    """
+
+    replay = True
+
+    def __init__(
+        self,
+        *,
+        run_id: str,
+        events: tuple,
+        scenario_name: str,
+        registry: Registry | None = None,
+        tools=None,
+    ) -> None:
+        self.run_id = run_id
+        self._events = tuple(events)
+        self._scenario_name = scenario_name
+        registry = registry or default_registry()
+        tools = tools if tools is not None else default_tool_registry()
+        scenario = registry.build_scenario(scenario_name, tools=tools)
+        self._scenario = scenario
+        self._cast = [agent.manifest for agent in scenario.agents]
+        self._governor = registry.governor_for(scenario_name)
+        obs.log("session.replay", scenario=scenario_name, run_id=run_id, events=len(self._events))
+
+    # ── read surface (mirrors FishbowlSession) ────────────────────────────────────
+
+    @property
+    def events(self):
+        return self._events
+
+    @property
+    def head(self) -> int:
+        return len(self._events)
+
+    @property
+    def scenario_name(self) -> str:
+        return self._scenario_name
+
+    @property
+    def goal(self) -> str:
+        return self._scenario.goal
+
+    @property
+    def cast(self) -> list[AgentManifest]:
+        return self._cast
+
+    def has_verdict(self) -> bool:
+        return any(e.kind == "judge.verdict" for e in self._events)
+
+    @property
+    def autoplay_tick_cap(self) -> int:
+        return self.head
+
+    # ── inert lifecycle (a replay never generates) ────────────────────────────────
+
+    def reset(self, *_args, **_kwargs) -> None:  # pragma: no cover - inert by design
+        return None
+
+    def step(self, *_args, **_kwargs) -> None:
+        return None
+
+    def step_one(self, *_args, **_kwargs) -> bool:
+        return False
+
+    def inject(self, *_args, **_kwargs) -> None:
+        return None
+
+    # ── snapshot ──────────────────────────────────────────────────────────────────
+
+    def snapshot(self, k: int | None = None) -> dict:
+        events = self._events
+        return view_model_at(
+            events,
+            k if k is not None else len(events),
+            self._cast,
+            scenario_name=self._scenario_name,
+            goal=self.goal,
+            governor=None,  # no live governor on a replay — meters show recorded text only
+            token_ceiling=getattr(self._governor, "max_total_tokens", None),
+            max_rounds=getattr(self._governor, "max_turns", None),
+        )

tests/test_fishbowl_archive.py

@@ -0,0 +1,244 @@
+"""Sessions & Archive: per-user run attribution, run-scoped live transcript, and the
+read-only replay the Lab's "Load" drawer hands to the Show.
+
+No mocks — the deterministic stub (conftest) drives the cast.  These tests use a
+*file-backed* SQLite DB (not the suite's default in-memory ``sqlite://``) because the
+Archive reads through a fresh ``make_ledger()`` each call, and only a file is shared
+across those connections — exactly how the real app's many sessions share one store.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.core.conductor import Conductor
+from src.core.events import Event, normalize_session_id
+from src.core.ledger_factory import make_ledger
+from src.core.manifest import AgentManifest, ScheduleConfig
+from src.core.registry import default_registry
+from src.scenarios.base import Scenario
+from src.ui.fishbowl import app as fb_app
+from src.ui.fishbowl.archive import list_runs, load_replay, run_card_label
+from src.ui.fishbowl.session import FishbowlSession, ReplaySession
+
+
+@pytest.fixture
+def shared_db(monkeypatch, tmp_path):
+    """Point every ``make_ledger()`` at one on-disk SQLite file (shared across sessions)."""
+    url = f"sqlite:///{tmp_path / 'events.db'}"
+    monkeypatch.setenv("DATABASE_URL", url)
+    return url
+
+
+def _two_scenarios() -> tuple[str, str]:
+    names = list(default_registry().scenarios)
+    assert len(names) >= 2, "need at least two scenarios to test cross-scenario scoping"
+    return names[0], names[1]
+
+
+def _run(scenario_name: str, session_id: str, *, seed: str = "seed") -> str:
+    """Start a run for *session_id* in *scenario_name*; return its run_id."""
+    session = FishbowlSession(scenario_name)
+    session.reset(seed, session_id=session_id)
+    return session.conductor.run_id
+
+
+# ── per-user attribution + scoping ────────────────────────────────────────────────
+
+
+class TestRunAttribution:
+    def test_run_started_carries_session_id(self, shared_db):
+        scenario, _ = _two_scenarios()
+        run_id = _run(scenario, "user-A")
+        started = next(e for e in make_ledger().events_for_run(run_id) if e.kind == "run.started")
+        assert started.payload.get("session_id") == "user-A"
+
+    def test_no_session_id_omits_the_key(self, shared_db):
+        scenario, _ = _two_scenarios()
+        session = FishbowlSession(scenario)
+        session.reset("seed")  # no session_id
+        started = next(e for e in make_ledger().events_for_run(session.conductor.run_id) if e.kind == "run.started")
+        assert "session_id" not in started.payload
+
+
+class TestLiveTranscriptIsRunScoped:
+    def test_session_events_never_bleed_other_runs(self, shared_db):
+        # The core fix: the ledger is one shared store of every run, but a live
+        # session must only ever see its own run — no cross-run/scenario bleed.
+        sx, sy = _two_scenarios()
+        a = FishbowlSession(sx)
+        a.reset("seed", session_id="user-A")
+        a_run = a.conductor.run_id
+
+        b = FishbowlSession(sy)  # a different world, same shared DB
+        b.reset("seed", session_id="user-A")
+
+        # b's ledger physically contains a's events (shared file)...
+        assert any(e.run_id == a_run for e in b.conductor.ledger.events)
+        # ...yet b's *view* shows only b's run, and a's only a's.
+        assert {e.run_id for e in b.events} == {b.conductor.run_id}
+        assert {e.run_id for e in a.events} == {a_run}
+
+
+# ── the Archive list ──────────────────────────────────────────────────────────────
+
+
+class TestListRuns:
+    def test_filters_to_my_runs_in_this_world(self, shared_db):
+        sx, sy = _two_scenarios()
+        mine_x = _run(sx, "me")
+        _run(sx, "someone-else")  # same world, different user
+        _run(sy, "me")  # my run, different world
+
+        rows = list_runs(sx, "me")
+        assert [r.run_id for r in rows] == [mine_x]
+        assert rows[0].session_id == "me"
+        assert rows[0].scenario == sx
+
+    def test_newest_first(self, shared_db):
+        sx, _ = _two_scenarios()
+        first = _run(sx, "me", seed="one")
+        second = _run(sx, "me", seed="two")
+        rows = list_runs(sx, "me")
+        assert [r.run_id for r in rows] == [second, first]
+
+    def test_empty_without_session_id(self, shared_db):
+        sx, _ = _two_scenarios()
+        _run(sx, "me")
+        assert list_runs(sx, None) == []
+        assert list_runs(sx, "") == []
+
+    def test_run_card_label_is_a_single_line(self, shared_db):
+        sx, _ = _two_scenarios()
+        _run(sx, "me")
+        label = run_card_label(list_runs(sx, "me")[0])
+        assert "\n" not in label and label.startswith("▶")
+
+
+# ── read-only replay ─────────────────────────────────────���─────────────────────────
+
+
+class TestReplay:
+    def test_load_replay_reconstructs_the_run(self, shared_db):
+        sx, _ = _two_scenarios()
+        run_id = _run(sx, "me")
+        expected = make_ledger().events_for_run(run_id)
+
+        replay = load_replay(run_id)
+        assert isinstance(replay, ReplaySession)
+        assert replay.replay is True
+        assert [e.id for e in replay.events] == [e.id for e in expected]
+        assert replay.scenario_name == sx
+        # The snapshot renders without a live conductor.
+        vm = replay.snapshot()
+        assert vm["total"] == len(expected)
+
+    def test_load_replay_unknown_run_is_none(self, shared_db):
+        assert load_replay("no-such-run") is None
+        assert load_replay("") is None
+
+    def test_replay_session_is_inert(self, shared_db):
+        sx, _ = _two_scenarios()
+        replay = load_replay(_run(sx, "me"))
+        head_before = replay.head
+        assert replay.step_one() is False
+        replay.step()
+        replay.inject("nudge")
+        assert replay.head == head_before  # nothing generated
+
+    def test_advance_replays_prefix_then_stops(self, shared_db):
+        sx, _ = _two_scenarios()
+        replay = load_replay(_run(sx, "me"))
+        assert replay.head >= 1
+        # From 0 it walks forward through the recorded prefix without generating.
+        k, ticks, reason = fb_app.advance_one_tick(replay, 0, 0)
+        assert (k, reason) == (1, None)
+        # At the head it stops cleanly (no token spend, no backstop trip).
+        k, ticks, reason = fb_app.advance_one_tick(replay, replay.head, 0)
+        assert k == replay.head and reason is not None
+
+
+# ── session hardening: envelope stamping, normalization, context scoping ──────────
+
+
+class _ProbeAgent:
+    """Records the run_ids of every event the conductor hands it as context."""
+
+    name = "probe"
+    manifest = AgentManifest(name="probe", persona="p", may_emit=["agent.spoke"], schedule=ScheduleConfig(tick_every=1))
+
+    def __init__(self) -> None:
+        self.last_usage: dict = {}
+        self.seen_run_ids: set[str] = set()
+
+    def act(self, run_id, turn, projection, recent_events) -> Event:
+        self.seen_run_ids.update(e.run_id for e in recent_events)
+        return Event(run_id=run_id, turn=turn, kind="agent.spoke", actor="probe", payload={"text": "t"})
+
+
+def _probe_conductor() -> tuple[Conductor, _ProbeAgent]:
+    probe = _ProbeAgent()
+    scenario = Scenario(name="probe-world", default_seed="seed", agents=(probe,))
+    return Conductor(scenario, ledger=make_ledger()), probe
+
+
+class TestNormalizeSessionId:
+    def test_accepts_minted_shapes(self):
+        assert normalize_session_id("3b456aab-4ac6-46d6-981a-ff10a8f25fdb") is not None
+        assert normalize_session_id("sess-abc123xyz") == "sess-abc123xyz"
+        assert normalize_session_id("  padded-id  ") == "padded-id"
+
+    def test_rejects_untrusted_garbage(self):
+        assert normalize_session_id(None) is None
+        assert normalize_session_id("") is None
+        assert normalize_session_id("a" * 65) is None  # over-long
+        assert normalize_session_id("has spaces") is None
+        assert normalize_session_id("<script>alert(1)</script>") is None
+
+    def test_conductor_normalizes_at_the_boundary(self, shared_db):
+        c, _probe = _probe_conductor()
+        c.reset("seed", session_id="<not a valid id>")
+        assert c.session_id is None
+        started = next(e for e in c.ledger.events_for_run(c.run_id) if e.kind == "run.started")
+        assert "session_id" not in started.payload and started.session_id is None
+
+
+class TestEnvelopeStamping:
+    def test_every_event_in_a_run_carries_the_session_id(self, shared_db):
+        c, _probe = _probe_conductor()
+        c.reset("seed", session_id="user-A")
+        c.step(2)  # genesis + agent turns → several event kinds
+        events = c.ledger.events_for_run(c.run_id)
+        assert len(events) >= 3
+        assert all(e.session_id == "user-A" for e in events)
+
+    def test_session_id_round_trips_through_sql(self, shared_db):
+        c, _probe = _probe_conductor()
+        c.reset("seed", session_id="user-A")
+        c.step(1)
+        # A fresh connection re-reads rows from disk — the envelope must survive.
+        reread = make_ledger().events_for_run(c.run_id)
+        assert reread and all(e.session_id == "user-A" for e in reread)
+
+    def test_unattributed_run_stays_null(self, shared_db):
+        c, _probe = _probe_conductor()
+        c.reset("seed")  # headless run, no session
+        c.step(1)
+        assert all(e.session_id is None for e in c.ledger.events_for_run(c.run_id))
+
+
+class TestAgentContextIsRunScoped:
+    def test_agents_never_see_other_runs_in_recent_events(self, shared_db):
+        # Run A fills the shared store with another user's discussion.
+        a, _ = _probe_conductor()
+        a.reset("seed", session_id="user-A")
+        a.step(2)
+        a_run = a.run_id
+
+        # Run B (fresh conductor, same DB) — its agent's context must be B-only.
+        b, probe_b = _probe_conductor()
+        b.reset("seed", session_id="user-B")
+        b.step(2)
+
+        assert any(e.run_id == a_run for e in b.ledger.events)  # store IS shared...
+        assert probe_b.seen_run_ids == {b.run_id}  # ...but the agent's context is not

tests/test_memory.py

@@ -83,7 +83,7 @@ class _RaisingIndex:
     def index(self, events):
         raise RuntimeError("backend down")
 
-    def search(self, query, k):
+    def search(self, query, k, run_id=None):
         raise RuntimeError("backend down")
 
 

tests/test_memory_index.py

@@ -49,15 +49,18 @@ class _FakeIndex:
     def __init__(self) -> None:
         self.store: dict[str, Event] = {}
         self.add_calls: list[str] = []
+        self.search_run_ids: list[str | None] = []
 
     def index(self, events: tuple[Event, ...]) -> None:
         for e in events:
             self.add_calls.append(e.id)
             self.store[e.id] = e  # upsert by id → idempotent
 
-    def search(self, query: str, k: int) -> list[Event]:
+    def search(self, query: str, k: int, run_id: str | None = None) -> list[Event]:
+        self.search_run_ids.append(run_id)
         q = set(query.lower().split())
-        scored = [(len(q & set(str(e.payload.get("text", "")).lower().split())), e) for e in self.store.values()]
+        pool = [e for e in self.store.values() if run_id is None or e.run_id == run_id]
+        scored = [(len(q & set(str(e.payload.get("text", "")).lower().split())), e) for e in pool]
         scored.sort(key=lambda t: t[0], reverse=True)
         return [e for _, e in scored[:k]]
 
@@ -115,6 +118,22 @@ class TestSalienceUsesIndex:
         mem.visible((mine, theirs, glob), current_turn=2, query="stage")
         assert set(idx.store) == {"mine", "glob"}  # 'theirs' never indexed
 
+    def test_search_is_scoped_to_the_candidates_run(self):
+        """The index spans every run in the shared store; recall must be scoped to
+        the run the candidates came from, so one show's (or one user's) discussion
+        never crowds another's relevance budget."""
+        idx = _FakeIndex()
+        # The index already holds an event from ANOTHER run that matches the query.
+        foreign = Event(run_id="other-run", turn=1, kind="world.observed", actor="n", payload={"text": "beacon glow"})
+        idx.index((foreign,))
+
+        ours = _event("world.observed", turn=2, text="beacon glow signal", eid="ours")  # run_id="r"
+        mem = SalienceMemory("x", top_k=2, index=idx)
+        recalled = mem.visible((ours,), current_turn=3, query="beacon glow")
+
+        assert idx.search_run_ids == ["r"]  # the search was run-scoped...
+        assert [e.id for e in recalled] == ["ours"]  # ...and the foreign hit never surfaced
+
     def test_recency_still_applies_with_index(self):
         """Relevance is one term; recency must still separate equally-relevant
         events so the index does not flatten the composite score."""

tests/test_run_history.py

@@ -0,0 +1,342 @@
+"""Run history & lifecycle: ``run.started`` enrichment, ``run.finished`` emission,
+non-destructive reset, the run-scoped read surface, and multi-run indexing.
+
+No mocks — the deterministic stub (conftest) and a couple of tiny in-process agents
+drive everything, mirroring the idiom in ``test_conductor.py``.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.core.conductor import Conductor
+from src.core.events import Event
+from src.core.governor import BudgetExceeded, Governor
+from src.core.ledger import Ledger
+from src.core.manifest import AgentManifest, ScheduleConfig
+from src.core.projections import rebuild_stage
+from src.core.run_index import RunSummary, index_runs, index_runs_from_ledger
+from src.core.sqlalchemy_ledger import SqlAlchemyLedger
+from src.scenarios.base import Scenario
+from src.ui.fishbowl.session import FishbowlSession
+
+
+# ── tiny in-process agents ──────────────────────────────────────────────────────
+
+
+class _JudgingAgent:
+    """Rules a verdict every tick — stands in for the live Judge so a run resolves."""
+
+    name = "judge"
+    manifest = AgentManifest(
+        name="judge",
+        persona="p",
+        may_emit=["judge.verdict"],
+        schedule=ScheduleConfig(tick_every=1),
+        model_endpoint="model://judge-endpoint",
+        model_profile="strong",
+    )
+
+    def __init__(self) -> None:
+        self.last_usage: dict = {}
+
+    def act(self, run_id, turn, projection, recent_events) -> Event:
+        return Event(
+            run_id=run_id,
+            turn=turn,
+            kind="judge.verdict",
+            actor="judge",
+            payload={"text": "the verdict", "winner": "judge"},
+        )
+
+
+class _SpeakingAgent:
+    name = "speaker"
+    manifest = AgentManifest(
+        name="speaker", persona="p", may_emit=["agent.spoke"], schedule=ScheduleConfig(tick_every=1)
+    )
+
+    def __init__(self) -> None:
+        self.last_usage: dict = {}
+
+    def act(self, run_id, turn, projection, recent_events) -> Event:
+        return Event(run_id=run_id, turn=turn, kind="agent.spoke", actor="speaker", payload={"text": "hi"})
+
+
+def _verdict_scenario() -> Scenario:
+    return Scenario(name="verdict-world", default_seed="seed", agents=(_JudgingAgent(),))
+
+
+# ── run.started enrichment ────────────────────────────────────────────────────────
+
+
+class TestRunStartedPayload:
+    def test_run_started_carries_cast_model_map_and_scenario(self):
+        c = Conductor(scenario=_verdict_scenario())
+        c.reset("seed")
+        started = next(e for e in c.ledger.events_for_run(c.run_id) if e.kind == "run.started")
+        assert started.payload["scenario"] == "verdict-world"
+        cast = started.payload["cast"]
+        assert cast["judge"] == {"model_endpoint": "model://judge-endpoint", "model_profile": "strong"}
+
+    def test_run_started_reports_unbound_agent_without_manifest_fields(self):
+        c = Conductor(scenario=Scenario(name="s", default_seed="seed", agents=(_SpeakingAgent(),)))
+        c.reset("seed")
+        started = next(e for e in c.ledger.events_for_run(c.run_id) if e.kind == "run.started")
+        assert started.payload["cast"]["speaker"]["model_endpoint"] is None
+
+
+# ── run.finished emission ─────────────────────────────────────────────────────────
+
+
+class TestRunFinished:
+    def test_finished_on_verdict_via_session_finalize(self):
+        session = FishbowlSession("thousand-token-wood")
+        session.reset()
+        # Drive until the live Judge rules (the stub cast resolves quickly).
+        for _ in range(session.autoplay_tick_cap):
+            if session.has_verdict():
+                break
+            session.step()
+        assert session.has_verdict(), "the stub cast 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
+        assert finished[0].payload["reason"] == "verdict"
+
+    def test_finished_on_budget_trip_carries_budget_reason(self):
+        # A per-turn cap of 1 trips BudgetExceeded on the SECOND agent; the conductor
+        # must close the run with reason "budget" before the stop propagates.
+        scenario = Scenario(name="s", default_seed="seed", agents=(_SpeakingAgent(), _SpeakingAgent()))
+        c = Conductor(scenario=scenario, governor=Governor(max_calls_per_turn=1))
+        c.reset("seed")
+        with pytest.raises(BudgetExceeded):
+            c.step()
+        finished = [e for e in c.ledger.events_for_run(c.run_id) if e.kind == "run.finished"]
+        assert len(finished) == 1
+        assert finished[0].payload["reason"] == "budget"
+
+    def test_finalize_is_idempotent(self):
+        c = Conductor(scenario=_verdict_scenario())
+        c.reset("seed")
+        first = c.finalize("user_stop")
+        second = c.finalize("verdict")  # must NOT emit a second run.finished
+        assert first is not None and second is not None
+        assert first.id == second.id
+        finished = [e for e in c.ledger.events_for_run(c.run_id) if e.kind == "run.finished"]
+        assert len(finished) == 1
+        assert finished[0].payload["reason"] == "user_stop"
+
+    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.
+        c = Conductor(scenario=scenario)
+        c.reset("seed")
+        c.step()  # judge rules: winner="judge"
+        # Reproduce FishbowlSession.finalize's derivation against this conductor.
+        run_events = c.ledger.events_for_run(c.run_id)
+        verdict = next(e for e in reversed(run_events) if e.kind == "judge.verdict")
+        winner = verdict.payload.get("winner")
+        started = next(e for e in run_events if e.kind == "run.started")
+        winning_model = (started.payload["cast"].get(winner) or {}).get("model_endpoint")
+        finished = c.finalize("verdict", winner=winner, winning_model=winning_model)
+        assert finished.payload["winner"] == "judge"
+        assert finished.payload["winning_model"] == "model://judge-endpoint"
+
+
+# ── non-destructive reset ─────────────────────────────────────────────────────────
+
+
+class TestNonDestructiveReset:
+    def test_reset_starts_new_run_and_keeps_prior_run_events(self):
+        c = Conductor(scenario=_verdict_scenario())
+        c.reset("seed-a")
+        old_run = c.run_id
+        c.step()
+        old_events = c.ledger.events_for_run(old_run)
+        assert old_events, "the first run produced events"
+
+        c.reset("seed-b")
+        new_run = c.run_id
+        assert new_run != old_run, "reset mints a NEW run_id"
+        # Prior run's events survive on the shared, append-only ledger.
+        assert c.ledger.events_for_run(old_run) == old_events
+        # The new run only sees its own run.started/genesis so far.
+        new_events = c.ledger.events_for_run(new_run)
+        assert new_events and all(e.run_id == new_run for e in new_events)
+        assert any(e.kind == "run.started" for e in new_events)
+
+
+# ── run-scoped read surface ───────────────────────────────────────────────────────
+
+
+def _multi_run_ledger(ledger: Ledger) -> dict[str, list[Event]]:
+    """Append two interleaved runs to *ledger*; return {run_id: events} oracle."""
+    expected: dict[str, list[Event]] = {"run-A": [], "run-B": []}
+    plan = [
+        ("run-A", 0, "run.started"),
+        ("run-B", 0, "run.started"),
+        ("run-A", 1, "agent.spoke"),
+        ("run-A", 1, "judge.verdict"),
+        ("run-B", 1, "agent.spoke"),
+        ("run-A", 2, "run.finished"),
+    ]
+    for run_id, turn, kind in plan:
+        payload = {"seed": run_id} if kind == "run.started" else {"text": "t"}
+        e = Event(run_id=run_id, turn=turn, kind=kind, actor="x", payload=payload)
+        ledger.append(e)
+        expected[run_id].append(e)
+    return expected
+
+
+class TestRunScopedReads:
+    @pytest.mark.parametrize("ledger_factory", [Ledger, lambda: SqlAlchemyLedger("sqlite://")])
+    def test_events_for_run_and_runs(self, ledger_factory):
+        ledger = ledger_factory()
+        expected = _multi_run_ledger(ledger)
+
+        # runs() reports distinct run_ids in first-seen order.
+        assert ledger.runs() == ("run-A", "run-B")
+
+        for run_id, events in expected.items():
+            got = ledger.events_for_run(run_id)
+            assert [e.id for e in got] == [e.id for e in events]
+            assert [e.kind for e in got] == [e.kind for e in events]
+
+    def test_rebuild_stage_ignores_other_runs(self):
+        ledger = Ledger()
+        _multi_run_ledger(ledger)
+        proj = rebuild_stage(ledger.events, "run-B")
+        # run-B never emitted a verdict; run-A did — scoping must not bleed it in.
+        assert proj.judge_notes == []
+
+    def test_rebuild_stage_unscoped_sees_everything(self):
+        ledger = Ledger()
+        _multi_run_ledger(ledger)
+        scoped = rebuild_stage(ledger.events, "run-B")
+        unscoped = rebuild_stage(ledger.events)
+        # The unscoped projection folds in run-A's verdict; the scoped one does not.
+        assert unscoped.judge_notes != []
+        assert scoped.judge_notes == []
+
+
+# ── run index: oracle vs. SQL implementation ──────────────────────────────────────
+
+
+def _index_oracle(events: tuple[Event, ...]) -> dict[str, list[str]]:
+    """Reference per-run index: {run_id: [event_id, ...]} in first-seen/offset order."""
+    index: dict[str, list[str]] = {}
+    for e in events:
+        index.setdefault(e.run_id, []).append(e.id)
+    return index
+
+
+class TestRunIndex:
+    def test_sql_run_index_matches_oracle(self):
+        sql = SqlAlchemyLedger("sqlite://")
+        mem = Ledger()
+        # Feed the SAME multi-run stream into a plain list to build the oracle.
+        plan = [
+            ("r1", "run.started"),
+            ("r2", "run.started"),
+            ("r1", "agent.spoke"),
+            ("r3", "run.started"),
+            ("r2", "agent.spoke"),
+            ("r1", "run.finished"),
+            ("r3", "agent.spoke"),
+            ("r2", "run.finished"),
+        ]
+        stream: list[Event] = []
+        for run_id, kind in plan:
+            e = Event(run_id=run_id, turn=0, kind=kind, actor="x", payload={"text": "t"})
+            sql.append(e)
+            mem.append(e)
+            stream.append(e)
+
+        oracle = _index_oracle(tuple(stream))
+
+        # runs() must agree with the oracle's first-seen run order.
+        assert list(sql.runs()) == list(oracle.keys())
+        assert list(mem.runs()) == list(oracle.keys())
+
+        # Per-run event ordering must match the oracle on the SQL backend.
+        for run_id, ids in oracle.items():
+            assert [e.id for e in sql.events_for_run(run_id)] == ids
+            assert [e.id for e in mem.events_for_run(run_id)] == ids
+
+
+# ── run_index module: index_runs oracle vs. index_runs_from_ledger (SQL path) ─────
+
+
+def _bookended_runs() -> list[Event]:
+    """A two-run stream with full run.started/run.finished payloads, interleaved."""
+    return [
+        Event(
+            run_id="r1",
+            turn=0,
+            kind="run.started",
+            actor="conductor",
+            payload={
+                "seed": "acorn",
+                "goal": "g",
+                "scenario": "thousand_token_wood",
+                "cast": {"judge": {"model_endpoint": "model://j", "model_profile": "strong"}},
+            },
+        ),
+        Event(
+            run_id="r2",
+            turn=0,
+            kind="run.started",
+            actor="conductor",
+            payload={"seed": "burrow", "scenario": "fishbowl", "cast": {}},
+        ),
+        Event(run_id="r1", turn=1, kind="agent.spoke", actor="judge", payload={"text": "t"}),
+        Event(
+            run_id="r1",
+            turn=2,
+            kind="run.finished",
+            actor="conductor",
+            payload={
+                "reason": "verdict",
+                "winner": "judge",
+                "winning_model": "model://j",
+                "turns": 2,
+                "tokens": 1234,
+            },
+        ),
+        # r2 starts but never finishes — its finished_* fields stay None/0.
+        Event(run_id="r2", turn=1, kind="agent.spoke", actor="x", payload={"text": "t"}),
+    ]
+
+
+class TestRunIndexModule:
+    def test_index_runs_folds_bookend_events(self):
+        summaries = index_runs(_bookended_runs())
+        assert [s.run_id for s in summaries] == ["r1", "r2"]  # first-seen order
+
+        r1, r2 = summaries
+        assert isinstance(r1, RunSummary)
+        assert (r1.scenario, r1.seed) == ("thousand_token_wood", "acorn")
+        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")
+        assert (r1.turns, r1.tokens) == (2, 1234)
+        assert r1.started_at is not None and r1.finished_at is not None
+
+        # r2 started but never finished — terminal fields stay at their defaults.
+        assert (r2.scenario, r2.seed) == ("fishbowl", "burrow")
+        assert r2.reason is None and r2.winner is None
+        assert (r2.turns, r2.tokens) == (0, 0)
+        assert r2.finished_at is None
+
+    @pytest.mark.parametrize("make_ledger", [Ledger, lambda: SqlAlchemyLedger("sqlite://")])
+    def test_from_ledger_matches_pure_oracle(self, make_ledger):
+        events = _bookended_runs()
+        ledger = make_ledger()
+        for e in events:
+            ledger.append(e)
+        # The indexed-query path must produce exactly what the pure oracle does.
+        assert index_runs_from_ledger(ledger) == index_runs(events)