Round 7: episodic memory experiment (per-scenario atoms)

Minimum-viable episodic memory layer per the user's "memory could be
tested" guidance — KB calls out a full hierarchical memory architecture
(working / episodic / semantic / graph / archival); this round ships
Episodic with strict gating, defers the rest.

Concept: at game completion, an LLM call extracts a few short
verified-against-verdicts insights ("memory atoms") from the debrief.
The selector keeps CONFIRMED + PLAUSIBLE atoms, drops SPECULATIVE,
dedupes >70% substring matches against existing atoms, and caps each
scenario at 50 atoms (oldest evicted). On the next game with the same
scenario, the top 3-5 most-recent atoms are prepended to write_briefing's
USER message under "## Memory of past games (treat as priors, not proof)"
— system prompt untouched so the prompt cache stays warm.

- wargame/memory/store.py — JSON per scenario at games_dir/memory/{id}.json
- wargame/memory/extractor.py — extract_atoms + selector + persistence
- wargame/engine/state.py — MemoryAtom, MemoryAtomList, MemoryConfidence
- wargame/engine/loop.py — extract_and_persist at game completion,
wrapped so memory failure never breaks a real game
- wargame/agents/control.py — _render_memory_block injects into briefing
user message; insights stay English regardless of game language so
atoms migrate cleanly across language switches
- wargame/cli.py — wargame memory list/clear/remove
- API: GET / DELETE /scenarios/{id}/memory[/{atom_id}]

321 tests pass (301 baseline + 20 new). CLAUDE.md §8 updated.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CLAUDE.md

@@ -229,7 +229,13 @@ in some paths.
   is per-individual-agent (one different model per role-agent) routing
   within a role — every agent that calls with `role="agent"` still shares
   one model in a given game.
-- German localization
+- German localization is shipped (per-game `language` field; `--language en|de`).
+- **Episodic memory** is shipped as a per-scenario atom-based store (see
+  `wargame/memory/`). The extractor runs on game completion, the selector
+  drops `SPECULATIVE` atoms and dedupes against the on-disk store, and the
+  control briefing prepends up to five PLAUSIBLE+ atoms as priors. The
+  hierarchical layers from the KB (Working, Semantic, Graph, Archival
+  memory) are still deferred.
 - Mid-game resume beyond simple snapshot reload
 - Replay viewer / WebSocket streaming of deliberations
 - Multiplayer

wargame/agents/control.py

@@ -10,6 +10,7 @@ Five responsibilities:
 
 from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING
 
 from wargame.agents.base import call_structured, current_language
@@ -20,12 +21,21 @@ from wargame.engine.state import (
     Briefing,
     CoachNote,
     Game,
+    MemoryAtom,
+    MemoryConfidence,
     Verdict,
     VerdictList,
     WorldState,
     WorldUpdate,
 )
 
+log = logging.getLogger(__name__)
+
+
+# How many atoms to inject per briefing. Capped low so the prompt stays
+# focused; the most-recent N (by `created_at`) win when more are on file.
+_MEMORY_INJECT_MAX = 5
+
 if TYPE_CHECKING:
     from wargame.agents.registry import ScenarioConfig
 
@@ -117,6 +127,59 @@ def _delta_summary(game: Game) -> str:
 # ---------------------------------------------------------------------------
 
 
+def _render_memory_block(atoms: list[MemoryAtom]) -> str:
+    """Render selected atoms into the prepend-to-user-message memory block.
+
+    Empty input → empty string (callers must NOT emit an empty heading).
+    Atoms are presented as priors, never as facts — the heading and the
+    block intro both say so explicitly so Control treats them as hints.
+    """
+    if not atoms:
+        return ""
+    lines = [
+        "## Memory of past games (treat as priors, not proof)",
+        "",
+        (
+            "These are insights distilled from previous games on this "
+            "scenario. They may be useful priors but should never override "
+            "the verdicts and world state from the current game."
+        ),
+        "",
+    ]
+    for atom in atoms:
+        provenance = f"source: game {atom.source_game_id[:8]}..."
+        if atom.source_turn is not None:
+            provenance += f", turn {atom.source_turn}"
+        lines.append(
+            f"- [confidence: {atom.confidence.value}] {atom.insight} "
+            f"({provenance})"
+        )
+    lines.append("")  # trailing blank line
+    return "\n".join(lines)
+
+
+def _fetch_memory_atoms_for_briefing(scenario_id: str) -> list[MemoryAtom]:
+    """Pull the top-N most-recent atoms for `scenario_id` (PLAUSIBLE+).
+
+    Wrapped in try/except so a corrupt memory file or import-time issue can
+    never break a briefing — episodic memory is auxiliary by construction.
+    """
+    try:
+        from wargame.memory import list_atoms
+
+        atoms = list_atoms(scenario_id, min_confidence=MemoryConfidence.PLAUSIBLE)
+    except Exception:  # noqa: BLE001
+        log.warning(
+            "Memory injection failed for scenario %s; briefing continues without it.",
+            scenario_id,
+            exc_info=True,
+        )
+        return []
+    # Most-recent first (the extractor records `created_at` in ISO 8601).
+    atoms.sort(key=lambda a: a.created_at, reverse=True)
+    return atoms[:_MEMORY_INJECT_MAX]
+
+
 def write_briefing(scenario: ScenarioConfig, game: Game) -> Briefing:
     world = game.world
     system = render(
@@ -127,7 +190,11 @@ def write_briefing(scenario: ScenarioConfig, game: Game) -> Briefing:
     # Per-turn volatile data goes in the user message so the system prefix
     # stays byte-identical across turns — a hard prerequisite for the
     # top-level prompt caching configured in `call_structured`.
+    memory_block = _render_memory_block(
+        _fetch_memory_atoms_for_briefing(scenario.scenario_id)
+    )
     user = (
+        f"{memory_block}"
         f"## Current world state\n\n"
         f"- Turn: **{world.turn} of {world.max_turns}**\n"
         f"- Public narrative: {world.public_narrative}\n"

wargame/api/routes.py

@@ -37,6 +37,8 @@ from wargame.api.schemas import (
     GameListResponse,
     HardConstraintInput,
     HardConstraintListResponse,
+    MemoryAtomListResponse,
+    MemoryClearResponse,
     MetricPoint,
     MetricsTimelinePoint,
     MetricsTimelineResponse,
@@ -1280,3 +1282,54 @@ def delete_constraint(
         game_id=game.id,
         constraints=list(game.hard_constraints),
     )
+
+
+# ---------------------------------------------------------------------------
+# Episodic memory endpoints (Round 7)
+#
+# Atoms are the per-scenario distillation of a finished game's lessons. The
+# read endpoint surfaces what's currently injected into briefings; the
+# delete endpoints let an operator wipe a single atom or the whole store
+# (useful while iterating on the extractor).
+# ---------------------------------------------------------------------------
+
+
+@router.get(
+    "/scenarios/{scenario_id}/memory",
+    response_model=MemoryAtomListResponse,
+)
+def get_memory(scenario_id: str) -> MemoryAtomListResponse:
+    """List PLAUSIBLE+ memory atoms for a scenario."""
+    from wargame.engine.state import MemoryConfidence
+    from wargame.memory import list_atoms
+
+    atoms = list_atoms(scenario_id, min_confidence=MemoryConfidence.PLAUSIBLE)
+    return MemoryAtomListResponse(scenario_id=scenario_id, atoms=atoms)
+
+
+@router.delete(
+    "/scenarios/{scenario_id}/memory/{atom_id}",
+    status_code=204,
+)
+def delete_memory_atom(scenario_id: str, atom_id: str):
+    """Remove a single memory atom by id. 404 if no such atom."""
+    from wargame.memory import remove_atom
+
+    if not remove_atom(scenario_id, atom_id):
+        raise HTTPException(
+            status_code=404,
+            detail=f"No atom {atom_id!r} on file for scenario {scenario_id!r}.",
+        )
+    return None
+
+
+@router.delete(
+    "/scenarios/{scenario_id}/memory",
+    response_model=MemoryClearResponse,
+)
+def delete_memory_all(scenario_id: str) -> MemoryClearResponse:
+    """Wipe every memory atom for a scenario."""
+    from wargame.memory import clear
+
+    cleared = clear(scenario_id)
+    return MemoryClearResponse(scenario_id=scenario_id, cleared=cleared)

wargame/api/schemas.py

@@ -21,6 +21,7 @@ from wargame.engine.state import (
     ConstraintSource,
     Game,
     HardConstraint,
+    MemoryAtom,
     PeekRecord,
     ReconResponse,
     RivalPlan,
@@ -485,3 +486,18 @@ class ActionCoachSessionResponse(BaseModel):
     synthesized_draft: ActionDraftResponse | None = None
     committed: bool = False
     messages: list[dict]  # raw dicts for clients to render the chat thread
+
+
+# ---------------------------------------------------------------------------
+# Episodic memory (Round 7)
+# ---------------------------------------------------------------------------
+
+
+class MemoryAtomListResponse(BaseModel):
+    scenario_id: str
+    atoms: list[MemoryAtom]
+
+
+class MemoryClearResponse(BaseModel):
+    scenario_id: str
+    cleared: int

wargame/cli.py

@@ -1639,5 +1639,90 @@ def cmd_resume(
     )
 
 
+# ---------------------------------------------------------------------------
+# Memory commands (Round 7)
+# ---------------------------------------------------------------------------
+
+
+memory_app = typer.Typer(
+    help="Inspect and manage episodic memory atoms persisted per scenario.",
+    no_args_is_help=True,
+)
+app.add_typer(memory_app, name="memory")
+
+
+@memory_app.command("list")
+def cmd_memory_list(
+    scenario_id: str = typer.Argument(..., help="Scenario id whose memory to read."),
+) -> None:
+    """List memory atoms for a scenario."""
+    from wargame.engine.state import MemoryConfidence
+    from wargame.memory import list_atoms
+
+    atoms = list_atoms(scenario_id, min_confidence=MemoryConfidence.PLAUSIBLE)
+    if not atoms:
+        console.print(
+            f"[dim]No atoms on file for scenario [bold]{scenario_id}[/bold].[/dim]"
+        )
+        return
+    t = Table(title=f"Memory atoms: {scenario_id}")
+    t.add_column("ID")
+    t.add_column("Confidence")
+    t.add_column("Insight")
+    t.add_column("Source game")
+    t.add_column("Turn", justify="right")
+    for atom in atoms:
+        truncated = atom.insight if len(atom.insight) <= 80 else atom.insight[:77] + "..."
+        t.add_row(
+            atom.id[:8] + "...",
+            atom.confidence.value,
+            truncated,
+            atom.source_game_id[:8] + "...",
+            str(atom.source_turn) if atom.source_turn is not None else "—",
+        )
+    console.print(t)
+
+
+@memory_app.command("clear")
+def cmd_memory_clear(
+    scenario_id: str = typer.Argument(..., help="Scenario id whose memory to wipe."),
+    yes: Annotated[
+        bool,
+        typer.Option("--yes", "-y", help="Skip confirmation prompt."),
+    ] = False,
+) -> None:
+    """Wipe every memory atom for a scenario."""
+    from wargame.memory import clear
+
+    if not yes:
+        confirm = typer.confirm(
+            f"Wipe ALL memory atoms for scenario {scenario_id!r}?",
+            default=False,
+        )
+        if not confirm:
+            console.print("[dim]Aborted.[/dim]")
+            raise typer.Exit(code=0)
+    removed = clear(scenario_id)
+    console.print(f"[green]Cleared {removed} atom(s) for {scenario_id}.[/green]")
+
+
+@memory_app.command("remove")
+def cmd_memory_remove(
+    scenario_id: str = typer.Argument(..., help="Scenario id."),
+    atom_id: str = typer.Argument(..., help="Atom id (full uuid)."),
+) -> None:
+    """Remove a single atom by id."""
+    from wargame.memory import remove_atom
+
+    removed = remove_atom(scenario_id, atom_id)
+    if removed:
+        console.print(f"[green]Removed atom {atom_id[:8]}... from {scenario_id}.[/green]")
+    else:
+        console.print(
+            f"[yellow]No atom {atom_id[:8]}... on file for {scenario_id}.[/yellow]"
+        )
+        raise typer.Exit(code=1)
+
+
 if __name__ == "__main__":
     app()

wargame/engine/loop.py

@@ -391,6 +391,28 @@ def advance(game: Game, *, generate_next_briefing: bool = True) -> Game:
             if current.turn >= game.world.max_turns:
                 game.status = "complete"
                 persistence.save(game)
+                # Episodic-memory extraction is best-effort and lives outside
+                # the player-visible flow. Wrapped in a defensive try/except so
+                # a malformed LLM response, a missing memory dir, or any other
+                # failure surfaces as a warning instead of breaking the debrief.
+                try:
+                    from wargame.memory import extractor as _memory_extractor
+
+                    persisted = _memory_extractor.extract_and_persist(scenario, game)
+                    if persisted:
+                        log.info(
+                            "Memory: persisted %d atom(s) for scenario %s "
+                            "(game %s).",
+                            len(persisted),
+                            scenario.scenario_id,
+                            game.id[:8],
+                        )
+                except Exception:  # noqa: BLE001
+                    log.warning(
+                        "Memory extraction failed for game %s; debrief unaffected.",
+                        game.id[:8],
+                        exc_info=True,
+                    )
                 return game
 
             # Start next turn (seed record, optionally write briefing).

wargame/engine/state.py

@@ -395,6 +395,49 @@ class HardConstraintList(BaseModel):
     constraints: list[HardConstraint] = Field(default_factory=list, max_length=20)
 
 
+# ---------------------------------------------------------------------------
+# Episodic memory — atoms extracted post-game and re-injected next game
+# ---------------------------------------------------------------------------
+
+
+class MemoryConfidence(str, Enum):
+    """Confidence label on a memory atom.
+
+    `CONFIRMED` insights are verdict-backed and not contradicted; `PLAUSIBLE`
+    insights are supported but partial; `SPECULATIVE` insights are filtered
+    out by the selector before persistence.
+    """
+
+    CONFIRMED = "confirmed"
+    PLAUSIBLE = "plausible"
+    SPECULATIVE = "speculative"
+
+
+class MemoryAtom(BaseModel):
+    """One transferable lesson from a finished game on a specific scenario.
+
+    Stored per-scenario in `{games_dir}/memory/{scenario_id}.json`. Insights
+    stay in English regardless of game language so atoms migrate cleanly when
+    the player switches languages between games on the same scenario.
+    """
+
+    id: str = Field(default_factory=_uuid)
+    scenario_id: str
+    insight: str = Field(min_length=10, max_length=400)
+    rationale: str = Field(min_length=5, max_length=400)
+    confidence: MemoryConfidence
+    source_game_id: str
+    source_turn: int | None = None
+    tags: list[str] = Field(default_factory=list, max_length=8)
+    created_at: str = Field(default_factory=_now)
+
+
+class MemoryAtomList(BaseModel):
+    """Wrapper for the extractor's structured LLM output."""
+
+    atoms: list[MemoryAtom] = Field(default_factory=list, max_length=20)
+
+
 # ---------------------------------------------------------------------------
 # World state
 # ---------------------------------------------------------------------------

wargame/memory/__init__.py

@@ -0,0 +1,25 @@
+"""Episodic memory: per-scenario atoms extracted from finished games.
+
+Atoms persist as JSON-per-scenario at `{games_dir}/memory/{scenario_id}.json`
+and are injected as priors into the briefing of the next game on the same
+scenario. The selector (in `extractor.py`) keeps only verdict-backed atoms
+to prevent self-degradation across runs.
+"""
+
+from __future__ import annotations
+
+from wargame.memory.store import (
+    add_atom,
+    clear,
+    list_atoms,
+    memory_path,
+    remove_atom,
+)
+
+__all__ = [
+    "add_atom",
+    "clear",
+    "list_atoms",
+    "memory_path",
+    "remove_atom",
+]

wargame/memory/extractor.py

@@ -0,0 +1,254 @@
+"""Post-game extractor that turns a finished game into MemoryAtom records.
+
+Single LLM call that reads the debrief markdown + verdict log + coach note
+and emits a `MemoryAtomList`. The selector then:
+
+1. Drops every `SPECULATIVE` atom (only verdict-backed insights survive).
+2. Dedupes against atoms already on disk for this scenario (case-insensitive
+   >70% substring overlap).
+3. Caps the per-scenario store at 50 atoms; oldest by `created_at` are
+   evicted when a new save would push the total over.
+
+Failure isolation: the engine wraps the entire extract+persist pipeline in
+try/except and only logs warnings, so a malformed LLM response never breaks
+the debrief flow.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING
+
+from wargame.agents.base import call_structured
+from wargame.debrief.exporter import export_debrief
+from wargame.engine.state import (
+    Game,
+    MemoryAtom,
+    MemoryAtomList,
+    MemoryConfidence,
+)
+from wargame.memory import store
+
+if TYPE_CHECKING:
+    from wargame.agents.registry import ScenarioConfig
+
+log = logging.getLogger(__name__)
+
+
+# Hard cap on persisted atoms per scenario. Keeps the briefing prompt
+# bounded even after dozens of plays. Oldest evicted by `created_at`.
+MAX_ATOMS_PER_SCENARIO = 50
+
+# Substring-overlap dedupe threshold. The selector drops a candidate atom
+# whose insight has more than this much character-level overlap with any
+# existing atom's insight (case-insensitive).
+DEDUP_OVERLAP_THRESHOLD = 0.70
+
+
+# ---------------------------------------------------------------------------
+# Extractor — single LLM call
+# ---------------------------------------------------------------------------
+
+
+def _extractor_system_prompt(world_bible: str) -> str:
+    """System prompt for the extractor.
+
+    NOTE: The extractor is the one place where the engine deliberately
+    overrides the per-game language directive. Atoms are stored in English
+    so a player who plays a scenario once in German and once in English
+    sees coherent shared memory across both runs — the briefing injector
+    simply renders the English insights inside whatever language the active
+    game was created in. The directive in the user message reinforces this.
+    """
+    return (
+        "You write episodic memory atoms after a finished war-gaming game. "
+        "Each atom captures one transferable lesson from THIS game that would "
+        "help a future CEO playing the SAME scenario. Be specific (named "
+        "entities, named amounts, named turns). Avoid generic platitudes "
+        "(\"communicate clearly\", \"manage stakeholders\"). The atom must "
+        "reference the world bible's named characters, companies, regulators, "
+        "or numbers.\n\n"
+        "## World bible (named entities, numbers, posture)\n\n"
+        f"{world_bible}\n\n"
+        "## Confidence rules — set per atom\n\n"
+        "- `confirmed` — a verdict in this game directly backs the insight "
+        "  (e.g., the rival's countermove fired exactly as the atom describes, "
+        "  with `effect_minor` or stronger).\n"
+        "- `plausible` — partial support in the verdicts and coach note; not "
+        "  a clean falsifiable observation but worth keeping as a prior.\n"
+        "- `speculative` — anything you can't tie to either a verdict outcome "
+        "  or the coach note's explicit framing. The engine drops these.\n\n"
+        "Each atom needs a `rationale` field that names the specific verdict "
+        "or coach passage that earned the confidence label. If you can't cite "
+        "evidence, label the atom `speculative`."
+    )
+
+
+def _extractor_user_prompt(scenario: ScenarioConfig, game: Game) -> str:
+    debrief_md = export_debrief(game)
+    coach_note_md = "_(no coach note on record)_"
+    if game.turns and game.turns[-1].verdicts:
+        # The coach note isn't on the TurnRecord; fall back to the debrief
+        # which already includes it (the debrief exporter calls write_coach_note).
+        pass
+    verdicts_log = json.dumps(
+        [
+            {
+                "turn": t.turn,
+                "verdicts": [v.model_dump() for v in t.verdicts],
+            }
+            for t in game.turns
+        ],
+        indent=2,
+        default=str,
+    )
+    # Deliberate exception to the per-game language directive: atoms stay
+    # English regardless of game language so they migrate cleanly across
+    # language switches between sessions on the same scenario.
+    english_override = (
+        "ALWAYS write the `insight` and `rationale` fields in clear English, "
+        "regardless of the language directive that may follow. Atoms are "
+        "stored shared across language settings. Proper nouns (Norhard "
+        "Thermal AG, Bettina Lange, Aergis Climate, …) stay verbatim."
+    )
+    return (
+        f"## Scenario: {scenario.scenario_id} — {scenario.company_name} vs "
+        f"{scenario.rival_name}\n\n"
+        f"## Source game id\n\n`{game.id}` (max turns: {game.world.max_turns})\n\n"
+        f"## Full debrief (already includes the coach note)\n\n{debrief_md}\n\n"
+        f"## Verdict log (turn-by-turn)\n\n{verdicts_log}\n\n"
+        f"## Output rules\n\n"
+        f"- Set `scenario_id` to `{scenario.scenario_id}` on every atom.\n"
+        f"- Set `source_game_id` to `{game.id}` on every atom.\n"
+        f"- Set `source_turn` to the turn the lesson was learned, when "
+        f"  the lesson is anchored to a specific turn; otherwise omit.\n"
+        f"- Emit at most 6 atoms — quality over quantity. One sharp atom "
+        f"  beats five vague ones.\n\n"
+        f"{english_override}\n\n"
+        f"Return a `MemoryAtomList` JSON object with the `atoms` field."
+    )
+
+
+def extract_atoms(scenario: ScenarioConfig, game: Game) -> MemoryAtomList:
+    """Single LLM call that produces candidate memory atoms from `game`.
+
+    Uses `role="evaluator"` so it inherits the per-role model override (the
+    Evaluator role tag was introduced in Round 3 specifically so calibration
+    workloads can route to a cheaper model).
+    """
+    system = _extractor_system_prompt(scenario.world_bible)
+    user = _extractor_user_prompt(scenario, game)
+    return call_structured(
+        schema=MemoryAtomList,
+        system=system,
+        user=user,
+        role="evaluator",
+        temperature=0.4,
+        max_tokens=4000,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Selector — gates promotion to long-term storage
+# ---------------------------------------------------------------------------
+
+
+def _normalise(text: str) -> str:
+    return " ".join(text.strip().casefold().split())
+
+
+def _overlap_ratio(a: str, b: str) -> float:
+    """Symmetric character-level overlap ratio.
+
+    Returns the longer-of-the-two substring-match ratio between `a` and `b`,
+    using `difflib.SequenceMatcher`. Cheap on the short strings we're
+    dealing with (atoms cap at 400 chars).
+    """
+    from difflib import SequenceMatcher
+
+    return SequenceMatcher(None, _normalise(a), _normalise(b)).ratio()
+
+
+def _is_duplicate(candidate: MemoryAtom, existing: list[MemoryAtom]) -> bool:
+    cand_norm = _normalise(candidate.insight)
+    for prev in existing:
+        prev_norm = _normalise(prev.insight)
+        # Substring containment counts as duplicate at any length — guards
+        # against an extractor that re-emits an existing atom with extra
+        # qualifying clauses bolted on either end.
+        if cand_norm in prev_norm or prev_norm in cand_norm:
+            return True
+        if _overlap_ratio(candidate.insight, prev.insight) > DEDUP_OVERLAP_THRESHOLD:
+            return True
+    return False
+
+
+def select_atoms(
+    candidates: MemoryAtomList,
+    *,
+    existing: list[MemoryAtom],
+) -> list[MemoryAtom]:
+    """Filter `candidates` against the selector rules.
+
+    Drops `SPECULATIVE` atoms, drops duplicates of `existing` atoms, and
+    drops new candidates that duplicate one another (so a single extractor
+    call that emits two near-identical atoms only persists one).
+    """
+    kept: list[MemoryAtom] = []
+    for atom in candidates.atoms:
+        if atom.confidence == MemoryConfidence.SPECULATIVE:
+            log.info(
+                "Memory selector: dropping speculative atom %r", atom.insight[:60]
+            )
+            continue
+        if _is_duplicate(atom, existing):
+            log.info(
+                "Memory selector: dropping duplicate atom %r", atom.insight[:60]
+            )
+            continue
+        if _is_duplicate(atom, kept):
+            log.info(
+                "Memory selector: dropping intra-batch duplicate %r",
+                atom.insight[:60],
+            )
+            continue
+        kept.append(atom)
+    return kept
+
+
+def _enforce_cap(scenario_id: str) -> None:
+    """Evict oldest atoms by `created_at` when over the per-scenario cap."""
+    atoms = store._read_atoms(scenario_id)  # noqa: SLF001 — package-private helper
+    if len(atoms) <= MAX_ATOMS_PER_SCENARIO:
+        return
+    atoms.sort(key=lambda a: a.created_at)
+    # Drop the oldest until we're back at the cap.
+    surviving = atoms[len(atoms) - MAX_ATOMS_PER_SCENARIO :]
+    store._write_atoms(scenario_id, surviving)  # noqa: SLF001
+    log.info(
+        "Memory: evicted %d oldest atoms for scenario %s to enforce 50-atom cap.",
+        len(atoms) - len(surviving),
+        scenario_id,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Public entry point — wired into `loop.advance` on game completion
+# ---------------------------------------------------------------------------
+
+
+def extract_and_persist(scenario: ScenarioConfig, game: Game) -> list[MemoryAtom]:
+    """Run the extractor + selector + persistence pipeline.
+
+    Idempotent in the sense that running it twice on the same finished game
+    just hits the dedupe path on the second call. Returns the atoms that
+    landed on disk.
+    """
+    candidates = extract_atoms(scenario, game)
+    existing = store._read_atoms(scenario.scenario_id)  # noqa: SLF001
+    selected = select_atoms(candidates, existing=existing)
+    for atom in selected:
+        store.add_atom(atom)
+    _enforce_cap(scenario.scenario_id)
+    return selected

wargame/memory/store.py

@@ -0,0 +1,124 @@
+"""JSON-per-scenario episodic memory store.
+
+One file per scenario at `{games_dir}/memory/{scenario_id}.json`, schema:
+`{"atoms": [MemoryAtom, ...]}`. The store does serialised read-modify-write
+file IO — sufficient for the single-CLI / single-uvicorn-worker MVP. If we
+later need true concurrent writers we'll layer a per-scenario file lock on
+top, but threading the same game across processes is out of scope today.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Iterable
+
+from pydantic import ValidationError
+
+from wargame.config import settings
+from wargame.engine.state import MemoryAtom, MemoryConfidence
+
+log = logging.getLogger(__name__)
+
+
+# Confidence ordering used to filter `list_atoms`: a query asking for the
+# minimum confidence `PLAUSIBLE` returns CONFIRMED + PLAUSIBLE atoms.
+_CONFIDENCE_RANK: dict[MemoryConfidence, int] = {
+    MemoryConfidence.SPECULATIVE: 0,
+    MemoryConfidence.PLAUSIBLE: 1,
+    MemoryConfidence.CONFIRMED: 2,
+}
+
+
+def _memory_dir() -> Path:
+    d = settings().games_dir / "memory"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def memory_path(scenario_id: str) -> Path:
+    """Resolve the JSON file backing memory for `scenario_id`."""
+    return _memory_dir() / f"{scenario_id}.json"
+
+
+def _read_atoms(scenario_id: str) -> list[MemoryAtom]:
+    path = memory_path(scenario_id)
+    if not path.exists():
+        return []
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError:
+        log.warning("Memory file for %s is corrupt; treating as empty.", scenario_id)
+        return []
+    out: list[MemoryAtom] = []
+    for item in raw.get("atoms", []):
+        try:
+            out.append(MemoryAtom.model_validate(item))
+        except ValidationError as e:
+            # Skip individually malformed atoms rather than failing the whole
+            # store — episodic memory is auxiliary, never load-bearing.
+            log.warning("Dropping malformed memory atom in %s: %s", scenario_id, e)
+    return out
+
+
+def _write_atoms(scenario_id: str, atoms: Iterable[MemoryAtom]) -> None:
+    path = memory_path(scenario_id)
+    payload = {"atoms": [a.model_dump() for a in atoms]}
+    path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+
+def list_atoms(
+    scenario_id: str,
+    *,
+    min_confidence: MemoryConfidence = MemoryConfidence.PLAUSIBLE,
+) -> list[MemoryAtom]:
+    """Return atoms for `scenario_id` at or above `min_confidence`.
+
+    Default `PLAUSIBLE` filter drops `SPECULATIVE` atoms — those should never
+    have made it past the selector, but keeping the filter here is the
+    defence-in-depth backstop against a corrupt JSON file.
+    """
+    threshold = _CONFIDENCE_RANK[min_confidence]
+    return [
+        a
+        for a in _read_atoms(scenario_id)
+        if _CONFIDENCE_RANK[a.confidence] >= threshold
+    ]
+
+
+def add_atom(atom: MemoryAtom) -> None:
+    """Append `atom` to the scenario's memory file. Dedupes on exact insight match.
+
+    The selector layer is responsible for the >70% substring dedupe; this
+    function only catches identical inserts so two extractor runs that
+    coincidentally produce the same string don't double-store.
+    """
+    existing = _read_atoms(atom.scenario_id)
+    insight_norm = atom.insight.strip().casefold()
+    for prev in existing:
+        if prev.insight.strip().casefold() == insight_norm:
+            return
+    existing.append(atom)
+    _write_atoms(atom.scenario_id, existing)
+
+
+def remove_atom(scenario_id: str, atom_id: str) -> bool:
+    """Remove the atom with `atom_id` if present; return True iff one was removed."""
+    existing = _read_atoms(scenario_id)
+    filtered = [a for a in existing if a.id != atom_id]
+    if len(filtered) == len(existing):
+        return False
+    _write_atoms(scenario_id, filtered)
+    return True
+
+
+def clear(scenario_id: str) -> int:
+    """Remove every atom for `scenario_id`. Returns the number removed."""
+    existing = _read_atoms(scenario_id)
+    if not existing:
+        # Still write an empty file so callers get a stable on-disk shape.
+        _write_atoms(scenario_id, [])
+        return 0
+    _write_atoms(scenario_id, [])
+    return len(existing)

wargame/tests/conftest.py

@@ -698,6 +698,51 @@ def _fake_hard_constraint_list() -> dict[str, Any]:
     }
 
 
+def _fake_memory_atom_list(user: str) -> dict[str, Any]:
+    """Two atoms — one CONFIRMED, one PLAUSIBLE — anchored to the fake
+    game's scenario id and game id. The user message embeds those values
+    via the extractor's prompt; we parse them back so the dispatched
+    fixture stays consistent across runs and across scenarios.
+    """
+    import re as _re
+
+    sid_match = _re.search(r"## Scenario:\s*([a-z0-9_]+)", user)
+    gid_match = _re.search(r"## Source game id\s*\n\s*\n`([0-9a-f]+)`", user)
+    scenario_id = sid_match.group(1) if sid_match else "competitor_disruption"
+    game_id = gid_match.group(1) if gid_match else "fake0000fake0000fake0000fake0000"
+    return {
+        "atoms": [
+            {
+                "scenario_id": scenario_id,
+                "insight": (
+                    "Bettina Lange the CFO will defect to a public margin-defense "
+                    "stance once installer-defection passes 7%."
+                ),
+                "rationale": (
+                    "Turn 2 verdict effect_minor on the CFO's deliberation "
+                    "matched this read; coach note flagged the alignment."
+                ),
+                "confidence": "confirmed",
+                "source_game_id": game_id,
+                "source_turn": 2,
+                "tags": ["cfo", "installer_channel"],
+            },
+            {
+                "scenario_id": scenario_id,
+                "insight": (
+                    "Aergis Climate counter-prices within 10 days when Norhard "
+                    "announces a rebate above EUR 5M scope."
+                ),
+                "rationale": "Aergis's revised plan in turns 3-4 implies this cadence.",
+                "confidence": "plausible",
+                "source_game_id": game_id,
+                "source_turn": 3,
+                "tags": ["aergis", "pricing"],
+            },
+        ]
+    }
+
+
 def _fake_scenario_map_synthesis() -> dict[str, Any]:
     return {
         "markdown": (
@@ -761,6 +806,8 @@ def _dispatch(schema_name: str, user: str) -> Any:
         return _fake_constraint_check_result()
     if schema_name == "HardConstraintList":
         return _fake_hard_constraint_list()
+    if schema_name == "MemoryAtomList":
+        return _fake_memory_atom_list(user)
     raise AssertionError(f"No fake canned for schema {schema_name!r}")
 
 
@@ -791,6 +838,7 @@ def fake_llm(monkeypatch: pytest.MonkeyPatch):
         "wargame.agents.critic",
         "wargame.agents.premortem",
         "wargame.agents.evaluator",
+        "wargame.memory.extractor",
         "wargame.scaffold.generator",
         "wargame.scaffold.setup_agent",
         "wargame.scaffold.extract_constraints",

wargame/tests/test_api.py

@@ -749,3 +749,84 @@ def test_sse_stream_route_is_registered(client: TestClient) -> None:
     app = client.app
     paths = {getattr(r, "path", None) for r in app.routes}
     assert "/games/{game_id}/stream" in paths
+
+
+# ---------------------------------------------------------------------------
+# Episodic memory endpoints (Round 7)
+# ---------------------------------------------------------------------------
+
+
+def test_get_memory_empty(client: TestClient) -> None:
+    r = client.get("/scenarios/competitor_disruption/memory")
+    assert r.status_code == 200
+    assert r.json() == {"scenario_id": "competitor_disruption", "atoms": []}
+
+
+def test_memory_populates_after_full_game(client: TestClient) -> None:
+    """Round-trip: complete a game via the API; the memory endpoint then lists atoms."""
+    action_body = {
+        "intent": "Counter-price selectively",
+        "justification": "Defend EBIT while stabilizing installer channel.",
+        "anchors": [
+            {"kind": "prior_analogue", "detail": "2018 KlimaPro price skirmish"},
+            {"kind": "stated_capability", "detail": "Finance can model in 48h"},
+            {"kind": "resource_committed", "detail": "EUR 2M margin buffer"},
+        ],
+    }
+    r = client.post("/games", json={"scenario_id": "competitor_disruption"})
+    gid = r.json()["game_id"]
+    for _ in range(5):
+        client.post(f"/games/{gid}/actions", json=action_body)
+        client.post(f"/games/{gid}/advance")
+
+    r = client.get("/scenarios/competitor_disruption/memory")
+    assert r.status_code == 200
+    atoms = r.json()["atoms"]
+    assert len(atoms) >= 1
+    assert any(a["confidence"] == "confirmed" for a in atoms)
+
+
+def test_delete_memory_atom_204(client: TestClient) -> None:
+    from wargame.engine.state import MemoryAtom, MemoryConfidence
+    from wargame.memory import add_atom, list_atoms
+
+    atom = MemoryAtom(
+        scenario_id="competitor_disruption",
+        insight="API delete test atom about Norhard's EBIT posture.",
+        rationale="Verdict-backed in turn 2.",
+        confidence=MemoryConfidence.CONFIRMED,
+        source_game_id="apitestgame00000",
+        source_turn=2,
+    )
+    add_atom(atom)
+    assert len(list_atoms("competitor_disruption")) == 1
+
+    r = client.delete(f"/scenarios/competitor_disruption/memory/{atom.id}")
+    assert r.status_code == 204
+    assert list_atoms("competitor_disruption") == []
+
+
+def test_delete_memory_atom_404(client: TestClient) -> None:
+    r = client.delete("/scenarios/competitor_disruption/memory/ghostid")
+    assert r.status_code == 404
+
+
+def test_delete_memory_all_returns_count(client: TestClient) -> None:
+    from wargame.engine.state import MemoryAtom, MemoryConfidence
+    from wargame.memory import add_atom
+
+    for i in range(3):
+        add_atom(
+            MemoryAtom(
+                scenario_id="competitor_disruption",
+                insight=f"Bulk delete test atom number {i} with enough chars.",
+                rationale="Verdict-backed in turn 2.",
+                confidence=MemoryConfidence.CONFIRMED,
+                source_game_id=f"apitest{i:08d}0",
+                source_turn=2,
+            )
+        )
+
+    r = client.delete("/scenarios/competitor_disruption/memory")
+    assert r.status_code == 200
+    assert r.json() == {"scenario_id": "competitor_disruption", "cleared": 3}

wargame/tests/test_memory.py

@@ -0,0 +1,331 @@
+"""Episodic memory tests — store round-trip, selector gating, briefing injection.
+
+The memory layer is auxiliary by design (a failed extractor must never break
+the debrief), so each test exercises one surface and asserts the failure
+isolation property where it applies.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from wargame.engine import loop
+from wargame.engine.state import (
+    ActionSubmission,
+    Anchor,
+    AnchorKind,
+    MemoryAtom,
+    MemoryAtomList,
+    MemoryConfidence,
+)
+
+
+def _player_action(intent: str = "Counter-price selectively") -> ActionSubmission:
+    return ActionSubmission(
+        actor="player",
+        intent=intent,
+        justification="Defend EBIT while stabilizing installer channel.",
+        anchors=[
+            Anchor(kind=AnchorKind.PRIOR_ANALOGUE, detail="2018 KlimaPro price skirmish"),
+            Anchor(kind=AnchorKind.STATED_CAPABILITY, detail="Finance can model in 48h"),
+            Anchor(kind=AnchorKind.RESOURCE_COMMITTED, detail="EUR 2M margin buffer"),
+        ],
+    )
+
+
+def _make_atom(
+    *,
+    insight: str,
+    confidence: MemoryConfidence = MemoryConfidence.CONFIRMED,
+    scenario_id: str = "competitor_disruption",
+    source_game_id: str = "fakegame00000000",
+    source_turn: int | None = 2,
+    created_at: str | None = None,
+) -> MemoryAtom:
+    atom = MemoryAtom(
+        scenario_id=scenario_id,
+        insight=insight,
+        rationale="Verdict on turn 2 backed this read.",
+        confidence=confidence,
+        source_game_id=source_game_id,
+        source_turn=source_turn,
+    )
+    if created_at is not None:
+        atom = atom.model_copy(update={"created_at": created_at})
+    return atom
+
+
+# ---------------------------------------------------------------------------
+# Store round-trip
+# ---------------------------------------------------------------------------
+
+
+class TestStoreRoundtrip:
+    def test_list_returns_empty_when_no_file(self, fake_llm) -> None:
+        from wargame.memory import list_atoms
+
+        assert list_atoms("competitor_disruption") == []
+
+    def test_add_then_list_returns_atom(self, fake_llm) -> None:
+        from wargame.memory import add_atom, list_atoms
+
+        atom = _make_atom(insight="Bettina Lange holds the EBIT line under 9%.")
+        add_atom(atom)
+        result = list_atoms("competitor_disruption")
+        assert len(result) == 1
+        assert result[0].id == atom.id
+        assert result[0].insight == atom.insight
+
+    def test_add_dedupes_exact_match(self, fake_llm) -> None:
+        from wargame.memory import add_atom, list_atoms
+
+        atom1 = _make_atom(insight="Identical insight string here is enough.")
+        # Two distinct atom instances with the same insight string.
+        atom2 = _make_atom(insight="Identical insight string here is enough.")
+        add_atom(atom1)
+        add_atom(atom2)
+        atoms = list_atoms("competitor_disruption")
+        assert len(atoms) == 1
+        assert atoms[0].id == atom1.id
+
+    def test_remove_returns_true_when_found(self, fake_llm) -> None:
+        from wargame.memory import add_atom, list_atoms, remove_atom
+
+        atom = _make_atom(insight="Aergis counter-prices within ten days when threatened.")
+        add_atom(atom)
+        assert remove_atom("competitor_disruption", atom.id) is True
+        assert list_atoms("competitor_disruption") == []
+
+    def test_remove_returns_false_when_missing(self, fake_llm) -> None:
+        from wargame.memory import remove_atom
+
+        assert remove_atom("competitor_disruption", "ghostid") is False
+
+    def test_clear_returns_count(self, fake_llm) -> None:
+        from wargame.memory import add_atom, clear, list_atoms
+
+        for i in range(3):
+            add_atom(_make_atom(insight=f"Insight number {i} carries enough text."))
+        cleared = clear("competitor_disruption")
+        assert cleared == 3
+        assert list_atoms("competitor_disruption") == []
+
+    def test_list_min_confidence_drops_speculative(self, fake_llm) -> None:
+        # Speculative atoms can land in the file via direct edits even though
+        # the selector normally filters them — assert the read filter holds.
+        from wargame.memory import store
+
+        confirmed = _make_atom(
+            insight="Confirmed insight that survives the read filter.",
+            confidence=MemoryConfidence.CONFIRMED,
+        )
+        speculative = _make_atom(
+            insight="Speculative insight that should be filtered out.",
+            confidence=MemoryConfidence.SPECULATIVE,
+        )
+        store._write_atoms("competitor_disruption", [confirmed, speculative])
+        from wargame.memory import list_atoms
+
+        atoms = list_atoms("competitor_disruption")
+        assert len(atoms) == 1
+        assert atoms[0].confidence == MemoryConfidence.CONFIRMED
+
+
+# ---------------------------------------------------------------------------
+# Selector
+# ---------------------------------------------------------------------------
+
+
+class TestSelector:
+    def test_selector_drops_speculative_keeps_confirmed_and_plausible(
+        self, fake_llm
+    ) -> None:
+        from wargame.memory.extractor import select_atoms
+
+        candidates = MemoryAtomList(
+            atoms=[
+                _make_atom(
+                    insight="Confirmed atom A about the CFO's posture.",
+                    confidence=MemoryConfidence.CONFIRMED,
+                ),
+                _make_atom(
+                    insight="Plausible atom B about Aergis's pricing cadence.",
+                    confidence=MemoryConfidence.PLAUSIBLE,
+                ),
+                _make_atom(
+                    insight="Speculative atom C with no real grounding here.",
+                    confidence=MemoryConfidence.SPECULATIVE,
+                ),
+            ]
+        )
+        kept = select_atoms(candidates, existing=[])
+        assert len(kept) == 2
+        assert all(a.confidence != MemoryConfidence.SPECULATIVE for a in kept)
+
+    def test_selector_dedupes_against_existing(self, fake_llm) -> None:
+        from wargame.memory.extractor import select_atoms
+
+        existing = [
+            _make_atom(
+                insight=(
+                    "Bettina Lange holds the EBIT line above 9% throughout "
+                    "the quarter when challenged."
+                )
+            )
+        ]
+        candidates = MemoryAtomList(
+            atoms=[
+                # Near-duplicate of the existing atom — should be dropped.
+                _make_atom(
+                    insight=(
+                        "Bettina Lange holds the EBIT line above 9% during "
+                        "the quarter when challenged."
+                    )
+                ),
+                # Genuinely different insight — should be kept.
+                _make_atom(
+                    insight=(
+                        "Aergis Climate accelerates its FSR-exposed PR moves "
+                        "when Norhard goes silent."
+                    )
+                ),
+            ]
+        )
+        kept = select_atoms(candidates, existing=existing)
+        assert len(kept) == 1
+        assert "Aergis Climate" in kept[0].insight
+
+    def test_selector_caps_at_50_evicting_oldest(self, fake_llm) -> None:
+        """Fill the store past 50 atoms and confirm the cap evicts oldest."""
+        from wargame.memory import list_atoms, store
+        from wargame.memory.extractor import _enforce_cap
+
+        # Seed 60 atoms with monotonically increasing created_at so the
+        # eviction order is deterministic.
+        atoms = []
+        for i in range(60):
+            atoms.append(
+                _make_atom(
+                    insight=f"Atom number {i:02d} with enough characters to validate.",
+                    created_at=f"2026-01-{(i % 28) + 1:02d}T00:{i:02d}:00+00:00",
+                )
+            )
+        # Sort to make the oldest deterministic.
+        atoms.sort(key=lambda a: a.created_at)
+        store._write_atoms("competitor_disruption", atoms)
+
+        _enforce_cap("competitor_disruption")
+        survived = list_atoms("competitor_disruption")
+        assert len(survived) == 50
+        # The oldest 10 should be gone.
+        survived_insights = {a.insight for a in survived}
+        for old in atoms[:10]:
+            assert old.insight not in survived_insights
+
+
+# ---------------------------------------------------------------------------
+# End-to-end: completing a game persists atoms; next game's briefing picks them up
+# ---------------------------------------------------------------------------
+
+
+def _play_full_game(scenario_id: str = "competitor_disruption") -> None:
+    """Run a full 5-turn game so the memory extractor fires on completion."""
+    game = loop.new_game(scenario_id)
+    for turn in range(1, game.world.max_turns + 1):
+        loop.act(game, _player_action(f"Turn {turn} action"))
+        game = loop.advance(game)
+    assert game.status == "complete"
+
+
+class TestEndToEnd:
+    def test_completing_a_game_persists_memory_atoms(self, fake_llm) -> None:
+        from wargame.memory import list_atoms
+
+        _play_full_game()
+        atoms = list_atoms("competitor_disruption")
+        assert len(atoms) >= 1
+        # The fake extractor emits one CONFIRMED and one PLAUSIBLE.
+        confidences = {a.confidence for a in atoms}
+        assert MemoryConfidence.CONFIRMED in confidences
+
+    def test_next_game_briefing_includes_memory_block(self, fake_llm) -> None:
+        from wargame.agents.control import _render_memory_block
+        from wargame.memory import list_atoms
+
+        _play_full_game()
+        # The injector's helper is the contract surface; ensure it produces
+        # a non-empty block when atoms are on file.
+        atoms = list_atoms("competitor_disruption")
+        block = _render_memory_block(atoms[:5])
+        assert "Memory of past games" in block
+        assert "treat as priors" in block.lower()
+        # Confidence label must appear inline so Control sees the gate.
+        assert "[confidence:" in block
+
+    def test_render_memory_block_empty_when_no_atoms(self, fake_llm) -> None:
+        from wargame.agents.control import _render_memory_block
+
+        assert _render_memory_block([]) == ""
+
+    def test_extractor_failure_does_not_break_debrief(
+        self, fake_llm, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Memory must NEVER break the player-visible flow."""
+        from wargame.memory import extractor
+
+        def _exploding_extractor(scenario, game):
+            raise RuntimeError("Simulated extractor failure")
+
+        monkeypatch.setattr(
+            extractor, "extract_and_persist", _exploding_extractor
+        )
+
+        # The full game still has to complete cleanly.
+        _play_full_game()
+
+
+# ---------------------------------------------------------------------------
+# Briefing system prompt is cache-stable across memory injections
+# ---------------------------------------------------------------------------
+
+
+def test_memory_injection_does_not_touch_system_prompt(
+    fake_llm, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Cache stability — atoms must land in the user message, not the system."""
+    from wargame.agents import control as control_mod
+    from wargame.memory import add_atom
+
+    captured: dict[str, str] = {}
+
+    real_call = control_mod.call_structured
+
+    def _capture(*, schema, system, user, role="agent", temperature=0.7, max_tokens=2000):
+        # Capture only the FIRST call — that's the briefing call we care about.
+        captured.setdefault("system", system)
+        captured.setdefault("user", user)
+        return real_call(
+            schema=schema,
+            system=system,
+            user=user,
+            role=role,
+            temperature=temperature,
+            max_tokens=max_tokens,
+        )
+
+    monkeypatch.setattr(control_mod, "call_structured", _capture)
+
+    # Seed a memory atom for the scenario before starting the game.
+    add_atom(
+        _make_atom(
+            insight=(
+                "Sentinel atom about Norhard Thermal AG that must not "
+                "land in the system prompt of the briefing."
+            )
+        )
+    )
+    loop.new_game("competitor_disruption")
+
+    assert "system" in captured and "user" in captured
+    assert "Sentinel atom about Norhard Thermal AG" not in captured["system"]
+    assert "Sentinel atom about Norhard Thermal AG" in captured["user"]