fix(engine): decontaminate recap/recall + harden persist_beat (F07-1, F14-3)

servers/engine/ledger.py

@@ -29,6 +29,34 @@
 
 KINDS = ("events", "dialogue", "decision", "npc_fact", "quest_milestone", "consequence", "lore")
 
+# F07-1 (issue #772): combat/system BOOKKEEPING must not enter the FTS index and
+# outrank story in recall (a recall('Rolan') probe returned 4 of 6 top hits as
+# bookkeeping). Two sources of contamination, decontaminated by EXACT discipline so the
+# documented DM-system-note path (SKILL.md:47 — a terse DM-authored kind=system note IS
+# meant to feed recall) is preserved:
+#   1. engine combat-event rows, stamped this schema in payload by _log_combat_event;
+#   2. the engine's two session markers ("Session N began" / "Session ended."), written
+#      by start_session/end_session — matched by exact prefix (same exact-match
+#      discipline as #749's wrapper-line filter), so a DM note that merely mentions a
+#      session is still indexed.
+_COMBAT_EVENT_SCHEMA = "clawdnd.combat_event.v1"
+# Anchored to the engine's own marker text; \b/(:|$) so a DM prose row that starts with
+# the same words but continues differently is NOT swallowed.
+_SESSION_MARKER_RE = re.compile(r"^(Session \d+ began\b|Session ended\.)")
+
+
+def _is_combat_event(e) -> bool:
+    """True iff a session-log entry is an engine combat-event bookkeeping row."""
+    if e.kind != "combat":
+        return False
+    payload = getattr(e, "payload", None)
+    return isinstance(payload, dict) and payload.get("schema") == _COMBAT_EVENT_SCHEMA
+
+
+def _is_session_marker(e) -> bool:
+    """True iff a kind=system row is one of the engine's two session markers."""
+    return e.kind == "system" and bool(_SESSION_MARKER_RE.match(e.text or ""))
+
 
 def _db_path(campaign_id: str):
     return store._campaign_dir(campaign_id) / "ledger.db"
@@ -193,6 +221,13 @@ def _ins(kind, text, who="", ref="", day=0):
                     # liveness filler, not campaign memory; recall must never surface it.
                     if is_wrapper_progress_line(e.text):
                         continue
+                    # F07-1 (#772): combat-event rows and the engine's session markers
+                    # are bookkeeping, not memory — skip them so recall ranks story, not
+                    # "Tough 1 takes 5 force damage" / "Session 2 began". A DM-authored
+                    # kind=system note (non-marker) still falls through and stays indexed
+                    # (SKILL.md:47 contract).
+                    if _is_combat_event(e) or _is_session_marker(e):
+                        continue
                     _ins("dialogue" if e.kind == "dialogue" else "events", e.text, who=e.speaker or "")
         for ch in campaign.characters.values():
             for fact in ch.memory:

servers/engine/recap.py

@@ -17,6 +17,25 @@
 # leave out of a "Previously on..." recap.
 _STORY_KINDS = frozenset({"narration", "dialogue", "combat"})
 
+# F07-1 (issue #772): a kind="combat" row can be EITHER a narrative beat (the DM's
+# prose: "the ogre roared") OR an engine bookkeeping row written by _log_combat_event,
+# which stamps this schema into payload. The mechanical rows ("Tough 1 takes 5 force
+# damage", "Turn advances to Tough 2") are not story — a "Previously on..." recap that
+# recites them reads as a damage log. We keep narrative combat (payload None or lacking
+# this schema) and drop only the schema-stamped rows. Distinct from #749/#763, which
+# exact-matched only the wrapper-progress heartbeat.
+_COMBAT_EVENT_SCHEMA = "clawdnd.combat_event.v1"
+
+
+def _is_combat_bookkeeping(entry: SessionLogEntry) -> bool:
+    """True iff this is an engine-authored combat-event row (schema-stamped payload),
+    i.e. mechanical bookkeeping rather than a narrative combat beat."""
+    if entry.kind != "combat":
+        return False
+    payload = entry.payload
+    return isinstance(payload, dict) and payload.get("schema") == _COMBAT_EVENT_SCHEMA
+
+
 _INTRO = "Previously on your adventure..."
 _EMPTY = "This is the start of a new adventure. The story has yet to be written."
 
@@ -56,9 +75,13 @@ def format_recap(entries: list[SessionLogEntry], max_entries: int = 12) -> str:
     # #749: the wrapper progress heartbeat ("Your move lands; attention gathers…") is a
     # liveness signal the QA/play wrappers log mid-turn, not story — reciting it in a
     # "Previously on…" recap reads as canned filler. Exact-match excluded.
+    # F07-1 (#772): schema-stamped combat-event rows are engine bookkeeping, not story —
+    # excluded here while narrative combat beats stay.
     story = [
         e for e in entries
-        if e.kind in _STORY_KINDS and not is_wrapper_progress_line(e.text)
+        if e.kind in _STORY_KINDS
+        and not is_wrapper_progress_line(e.text)
+        and not _is_combat_bookkeeping(e)
     ]
     recent = story[-max_entries:]
 

servers/engine/server.py

@@ -9685,9 +9685,12 @@ def persist_beat(
                        already streamed re-logs it twice. Leave events empty unless
                        you have a record row you did NOT already log live.
       - ``memories`` — list of facts to append, each
-                       ``{"character_id","fact"}``. Same as remember (de-duped per
-                       character). Target the COMPANION's id after a character beat
-                       AND the PC's id for what the hero learns — symmetric memory.
+                       ``{"character_id","fact"}`` (``character_id`` also accepts the
+                       ``id``/``npc_id`` aliases and is resolved tolerantly — a slug or
+                       a name finds the character, an unknown id raises with a
+                       did-you-mean). Same as remember (de-duped per character). Target
+                       the COMPANION's id after a character beat AND the PC's id for what
+                       the hero learns — symmetric memory.
       - ``decision`` — a single dict ``{"summary", "options"?, "chosen"?,
                        "rationale"?, "actor_ids"?, "sets_flag"?}``. Same as
                        record_decision (records the choice; sets_flag arms a gated
@@ -9699,7 +9702,13 @@ def persist_beat(
                        long_rest — those are their own beats, not a persist step.
 
     Returns a per-section summary: ``{"logged":[...], "remembered":[...],
-    "decision":{...}|None, "time":{...}|None}``.
+    "decision":{...}|None, "time":{...}|None}``. Each ``remembered`` row is the slim
+    ``{"id","fact","memory_count"}`` (the applied fact + the character's new fact count),
+    not the whole memory list.
+
+    The whole batch is VALIDATED before the first write: an unresolvable memories id, a
+    text-less events item, or a bad decision raises BEFORE any session-log row is
+    appended, so a failed call leaves no partial chronicle (retry-safe).
     """
     logged: list[dict] = []
     remembered: list[dict] = []
@@ -9710,46 +9719,102 @@ def persist_beat(
     # lock+load+fsync-save per write. (advance_time is handled AFTER, as its own
     # locked call, because its body — worldsim ticks, effect expiry, combat guard —
     # is non-trivial and re-entering campaign_lock here would deadlock.)
+    #
+    # F14-3 (#795): VALIDATE-THEN-APPLY. _log_session_entry writes the session jsonl
+    # IMMEDIATELY (append_log -> disk), so the old apply-and-validate interleave left a
+    # crash mid-batch with the events leg already on disk and the rest dropped — a retry
+    # then duplicated the chronicle rows. The non-atomic window is EVENTS ONLY (memories/
+    # decision mutate the in-memory snapshot and persist only at the block-end save, so a
+    # raise discards them). So we resolve EVERY item — coalesce event text + reject empty,
+    # resolve every memories character_id via the _char resolver (#786, F14-8) with id
+    # aliases, build the Decision with null-coerced str fields — BEFORE the first
+    # append_log. Any failure now precedes the first write -> atomic-in-effect, retry-safe.
     if events or memories or decision:
         with campaign_lock(campaign_id):
             c = _require(campaign_id)
-            for ev in (events or []):
+
+            # ---- PHASE 1: validate the whole batch (no writes) ----
+            planned_events: list[dict] = []
+            for i, ev in enumerate(events or []):
                 if not isinstance(ev, dict):
-                    raise ValueError("each events item must be a dict {kind,text,...}")
-                entry = _log_session_entry(
-                    c,
-                    kind=ev.get("kind", "narration"),
-                    text=ev.get("text", ""),
-                    speaker=ev.get("speaker") or "",
-                    payload=ev.get("payload"),
-                )
-                logged.append(entry.model_dump())
-            for mem in (memories or []):
+                    raise ValueError(f"events index {i}: each item must be a dict {{kind,text,...}}")
+                # log_event's alias set: text | message | content | note (text wins).
+                text = ev.get("text") or ev.get("message") or ev.get("content") or ev.get("note") or ""
+                if not text:
+                    raise ValueError(
+                        f"events index {i}: needs text (pass `text` or an alias: "
+                        f"`message`/`content`/`note`)"
+                    )
+                planned_events.append({
+                    "kind": ev.get("kind") or "narration",
+                    "text": text,
+                    "speaker": ev.get("speaker") or "",
+                    "payload": ev.get("payload"),
+                })
+
+            planned_memories: list[tuple] = []  # (Character, fact)
+            for i, mem in enumerate(memories or []):
                 if not isinstance(mem, dict):
-                    raise ValueError("each memories item must be a dict {character_id,fact}")
-                ch = _char(c, mem["character_id"])
-                fact = mem.get("fact", "")
-                if fact and fact not in ch.memory:  # de-dupe identical facts (matches remember)
-                    ch.memory.append(fact)
-                remembered.append({"id": ch.id, "name": ch.name, "memory": ch.memory})
+                    raise ValueError(f"memories index {i}: each item must be a dict {{character_id,fact}}")
+                # Accept character_id or the id/npc_id aliases the top-level tools tolerate,
+                # instead of a bare KeyError ('character_id') — the worst string on the surface.
+                cid_in = mem.get("character_id") or mem.get("id") or mem.get("npc_id")
+                if not cid_in:
+                    raise ValueError(
+                        f"memories index {i}: missing character_id "
+                        f"(pass `character_id`, or the alias `id`/`npc_id`)"
+                    )
+                try:
+                    ch = _char(c, cid_in)  # resolve-then-suggest (raises ValueError w/ did-you-mean)
+                except ValueError as e:
+                    raise ValueError(f"memories index {i}: {e}") from None
+                fact = mem.get("fact") or ""
+                planned_memories.append((ch, fact))
+
+            planned_decision: Optional[Decision] = None
+            decision_flag = ""
             if decision:
                 if not isinstance(decision, dict):
                     raise ValueError("decision must be a dict {summary,...}")
-                d = Decision(
+                # None-coerce every str field: the DM legitimately passes chosen=null for a
+                # still-open decision; `.get(k, "")` only defaults a MISSING key, an explicit
+                # null still reaches pydantic's str field and string_type-crashes the batch.
+                planned_decision = Decision(
                     day=c.day,
-                    summary=decision.get("summary", ""),
+                    summary=decision.get("summary") or "",
                     options=list(decision.get("options") or []),
-                    chosen=decision.get("chosen", ""),
-                    rationale=decision.get("rationale", ""),
+                    chosen=decision.get("chosen") or "",
+                    rationale=decision.get("rationale") or "",
                     actor_ids=list(decision.get("actor_ids") or []),
                 )
-                c.decisions.append(d)
-                flag = str(decision.get("sets_flag", "") or "").strip()
-                if flag:
-                    c.flags[flag] = True  # content-defined; arms a matching agenda's decision_flag
-                decision_out = {"id": d.id, "summary": d.summary, "chosen": d.chosen, "day": d.day}
-                if flag:
-                    decision_out["flag"] = flag
+                decision_flag = str(decision.get("sets_flag") or "").strip()
+
+            # ---- PHASE 2: apply (every item validated; first write is here) ----
+            for pe in planned_events:
+                entry = _log_session_entry(
+                    c,
+                    kind=pe["kind"],
+                    text=pe["text"],
+                    speaker=pe["speaker"],
+                    payload=pe["payload"],
+                )
+                logged.append(entry.model_dump())
+            for ch, fact in planned_memories:
+                if fact and fact not in ch.memory:  # de-dupe identical facts (matches remember)
+                    ch.memory.append(fact)
+                # Slim row (#795): the FACT just applied + a count, NOT the whole growing
+                # memory list per item (the old O(items x memory) quadratic echo).
+                remembered.append({"id": ch.id, "fact": fact, "memory_count": len(ch.memory)})
+            if planned_decision is not None:
+                c.decisions.append(planned_decision)
+                if decision_flag:
+                    c.flags[decision_flag] = True  # content-defined; arms a matching agenda's decision_flag
+                decision_out = {
+                    "id": planned_decision.id, "summary": planned_decision.summary,
+                    "chosen": planned_decision.chosen, "day": planned_decision.day,
+                }
+                if decision_flag:
+                    decision_out["flag"] = decision_flag
             save_campaign(c)  # ONE atomic write for all of the above
 
     # advance_time as its own locked call (sequential, not nested → no deadlock).

servers/engine/tests/test_beat_roundtrip.py

@@ -386,3 +386,107 @@ def test_persist_beat_does_not_advance_clock_during_combat(cid):
     assert out["time"]["phases_advanced"] == 0  # guarded
     after = store.load_campaign(cid)
     assert (after.day, after.time_of_day) == (before.day, before.time_of_day)
+
+
+# ── F14-3 (issue #795): validate-before-write, chosen:null guard, no bare KeyError ──
+
+
+def _session_log_lines(cid: str) -> list:
+    """Every session-log entry across the campaign (to assert atomic non-application)."""
+    c = store.load_campaign(cid)
+    entries = []
+    for sid in c.session_ids:
+        entries.extend(store.read_log(cid, sid))
+    return entries
+
+
+def test_persist_beat_chosen_null_does_not_crash(cid):
+    # The DM legitimately records a still-open decision with chosen=null. It must
+    # succeed (coerced to "") instead of pydantic string_type-crashing the batch.
+    out = server.persist_beat(
+        cid,
+        decision={"summary": "Trust the broker or walk", "chosen": None},
+    )
+    assert out["decision"] is not None
+    assert out["decision"]["chosen"] == ""
+    after = store.load_campaign(cid)
+    assert after.decisions[-1].chosen == ""
+
+
+def test_persist_beat_decision_null_str_fields_coerced(cid):
+    # summary / rationale passed as null must coerce, not crash (same latent class).
+    out = server.persist_beat(
+        cid,
+        decision={"summary": None, "options": None, "chosen": None,
+                  "rationale": None, "actor_ids": None},
+    )
+    assert out["decision"] is not None
+    after = store.load_campaign(cid)
+    d = after.decisions[-1]
+    assert d.summary == "" and d.chosen == "" and d.rationale == ""
+
+
+def test_persist_beat_bad_memory_id_is_actionable_not_bare_keyerror(cid):
+    # A mis-keyed / unknown memories character_id must yield an ACTIONABLE error
+    # (names the section + index + did-you-mean), never a bare KeyError 'character_id'.
+    with pytest.raises(Exception) as ei:
+        server.persist_beat(cid, memories=[{"fact": "no id here"}])
+    msg = str(ei.value)
+    assert "character_id" not in msg or "memories" in msg  # not the bare KeyError string
+    assert "memories" in msg and ("index 0" in msg or "[0]" in msg)
+
+
+def test_persist_beat_events_not_applied_when_later_section_fails(cid):
+    # ATOMICITY (events-only window, F14-3): a good event + a bad memory item must
+    # leave ZERO new session-log rows — validation precedes the first append_log.
+    char = _a_char(cid)
+    before = len(_session_log_lines(cid))
+    with pytest.raises(Exception):
+        server.persist_beat(
+            cid,
+            events=[{"kind": "narration", "text": "This line must NOT persist."}],
+            memories=[{"character_id": "no-such-id", "fact": "x"}],
+        )
+    after = len(_session_log_lines(cid))
+    assert after == before  # the events leg was NOT applied
+
+
+def test_persist_beat_event_text_alias_honored(cid):
+    # An events item keyed `message` (log_event's alias) must log the text, not empty.
+    out = server.persist_beat(
+        cid, events=[{"kind": "narration", "message": "Alias text lands."}]
+    )
+    assert out["logged"][0]["text"] == "Alias text lands."
+
+
+def test_persist_beat_event_all_text_aliases_missing_is_rejected(cid):
+    # An events item with no text under any alias must be REJECTED (not empty-logged).
+    before = len(_session_log_lines(cid))
+    with pytest.raises(Exception):
+        server.persist_beat(cid, events=[{"kind": "narration"}])
+    assert len(_session_log_lines(cid)) == before  # nothing written
+
+
+def test_persist_beat_memory_id_alias_resolves(cid):
+    # A memories item keyed `id` (instead of character_id) must resolve, not KeyError.
+    char = _a_char(cid)
+    out = server.persist_beat(cid, memories=[{"id": char, "fact": "Reached via id alias."}])
+    assert out["remembered"][0]["id"] == char
+    assert "Reached via id alias." in store.load_campaign(cid).characters[char].memory
+
+
+def test_persist_beat_remembered_return_is_not_quadratic(cid):
+    # 4 facts for one character must return O(items) rows carrying the FACT + a count,
+    # NOT the whole growing memory list per item (the quadratic echo).
+    char = _a_char(cid)
+    facts = [f"Fact number {i}." for i in range(4)]
+    out = server.persist_beat(
+        cid, memories=[{"character_id": char, "fact": f} for f in facts]
+    )
+    assert len(out["remembered"]) == 4
+    for row in out["remembered"]:
+        assert "memory" not in row  # no embedded growing list
+        assert row["id"] == char
+        assert "fact" in row and "memory_count" in row
+    # the per-item fact echoes back, not the whole list
+    assert [r["fact"] for r in out["remembered"]] == facts