fix(engine): load_slot rolls back session logs (F08-1) + lookup_lore reachable for natural queries (F10-1)

servers/engine/lorebook.py

@@ -87,10 +87,35 @@ def has_corpus(world_id: str) -> bool:
     return bool(_pages(world_id))
 
 
+# F10-1 — tier-0 noise floor. An authored (tier-0) page keeps absolute precedence over the wiki
+# tier only if its FTS |rank| is at least this FRACTION of the best-matching page overall — i.e.
+# it must match RELATIVE to the strongest hit this query produced. The floor is intentionally
+# relative (not a fixed absolute epsilon): bm25 |rank| magnitudes are corpus-size-dependent (in a
+# 2-page corpus every match ranks ≈ 2e-6; in the 356-page shipped corpus genuine matches rank
+# ≈ 5.8), so an absolute cutoff would wrongly demote a genuine authored page in a tiny corpus.
+# The fraction is small (0.01) so it ONLY catches near-zero noise: a pure-stopword match ranks
+# ~6 ORDERS OF MAGNITUDE below a real match (the "the Counting House" repro: noise ≈ 2e-6 vs best
+# ≈ 8.97 → ratio ~3e-7 ≪ 0.01, demoted), while a genuine-but-modest authored match (≈ 5.8, ratio
+# ~0.65) is comfortably kept. best == 0 (every page matched every token) → floor 0 → no demotion
+# → today's behavior.
+_NOISE_FLOOR_FRACTION = 0.01
+
+
 def _safe_match(query: str) -> str:
-    """OR-of-quoted-tokens (relevance, not all-terms-required) — same fix as recall."""
+    """OR-of-quoted-tokens (relevance, not all-terms-required) — same fix as recall.
+
+    F10-1: DROP sub-2-char tokens. A possessive query like ``"Wyrm's Crossing"`` tokenizes to
+    ``["Wyrm", "s", "Crossing"]``; the bare 1-char ``"s"`` matches the apostrophe-s of EVERY
+    page and drags unrelated pages above the dedicated one (the possessive-query failure class).
+    Single letters carry no retrieval signal, so we drop ``len < 2`` tokens — UNLESS that would
+    empty the match (a degenerate all-short query), in which case we keep them so the search
+    still runs rather than silently returning nothing. A query with no word-chars stays empty,
+    exactly as before."""
     toks = re.findall(r"[A-Za-z0-9]+", query or "")
-    return " OR ".join(f'"{t}"' for t in toks)
+    kept = [t for t in toks if len(t) >= 2]
+    if not kept:
+        kept = toks  # all tokens were short — don't empty the match; search them as-is
+    return " OR ".join(f'"{t}"' for t in kept)
 
 
 def _excerpt(text: str, tokens: list[str], width: int = 600) -> str:
@@ -201,26 +226,48 @@ def lookup_lore(
             [(i, p["title"], p["text"], p["source"], p.get("tier", 1)) for i, p in enumerate(pages)],
         )
 
-        def _match_tier(tier: int, n: int) -> list[int]:
+        def _match_tier(tier: int, n: int) -> list[tuple[int, float]]:
             # Filter on the UNINDEXED tier column alongside MATCH so authored matches
             # are found regardless of how many wiki pages also match (a bm25 over-fetch
             # over a 250-page corpus would otherwise bury the few short authored pages).
+            # Returns (rowid, rank) so the caller can apply the tier-0 NOISE FLOOR (F10-1):
+            # FTS `rank` is negative; a smaller |rank| ≈ a weaker match (≈0 == matched only a
+            # stopword), and we use |rank| to decide whether a tier-0 hit genuinely matches.
             try:
-                return [r[0] for r in conn.execute(
-                    "SELECT rowid FROM lore WHERE lore MATCH ? AND tier = ? ORDER BY rank LIMIT ?",
+                return [(r[0], r[1]) for r in conn.execute(
+                    "SELECT rowid, rank FROM lore WHERE lore MATCH ? AND tier = ? ORDER BY rank LIMIT ?",
                     (match, tier, n),
                 ).fetchall()]
             except sqlite3.OperationalError:
                 return []
 
         cap = max(limit, 1)
-        # Over-fetch each tier so the de-confliction can demote/drop contradicting hits
-        # and still fill `cap` with clean ones (without it, dropping a top hit would just
-        # shrink the result instead of promoting the next clean page).
-        fetch = cap * 3 if supersedes else cap
-        ids = _match_tier(0, fetch) + _match_tier(1, fetch)  # authored canon first, then wiki to fill
+        # Over-fetch each tier so the de-confliction (and the F10-1 noise-floor demotion below)
+        # can drop/demote a hit and still fill `cap` with clean ones — without it, dropping a
+        # top hit would just shrink the result instead of promoting the next clean page. (The
+        # noise floor needs the over-fetch too: a demoted noise-rank tier-0 page must be able to
+        # be replaced by a genuinely-matching tier-1 page that the old `fetch=cap` never read.)
+        fetch = cap * 3
+        t0 = _match_tier(0, fetch)  # authored canon
+        t1 = _match_tier(1, fetch)  # ingested wiki
     finally:
         conn.close()
+
+    # F10-1 — tier-0 NOISE FLOOR (tighten the authored-canon precedence to GENUINE matches).
+    # The tier-0-first guarantee exists so a short authored page isn't bm25-buried by the
+    # 351-page wiki tier (the post-canon de-confliction guard). But a stopword-heavy query
+    # ("the Counting House") makes a few authored pages match at NOISE rank (≈0, on "the"
+    # alone), and the old absolute precedence let those noise matches fill the cap and bury the
+    # dedicated wiki page. We keep tier-0-FIRST only for authored pages whose |rank| clears a
+    # noise floor — within a bounded fraction of the best-matching page overall, or a tiny
+    # absolute epsilon — and DEMOTE the rest below the wiki tier (kept as a fallback, never
+    # dropped). A clean query (every tier-0 match is genuine) leaves `weak0` empty, so the order
+    # reduces to today's `t0 + t1` and the output is byte-identical.
+    best_overall = max((abs(r) for _, r in (t0 + t1)), default=0.0)
+    floor = _NOISE_FLOOR_FRACTION * best_overall
+    strong0 = [rid for rid, rank in t0 if abs(rank) >= floor]
+    weak0 = [rid for rid, rank in t0 if abs(rank) < floor]
+    ids = strong0 + [rid for rid, _ in t1] + weak0  # genuine authored, then wiki, then noise authored
     tokens = re.findall(r"[A-Za-z0-9]+", query or "")
     subs = [s.lower() for s in (supersedes or []) if str(s).strip()]
 

servers/engine/store.py

@@ -180,31 +180,132 @@ def _slot_path(campaign_id: str, slot: str) -> Path:
     return _slots_dir(campaign_id) / f"{safe_path_segment(slot, 'slot')}.json"
 
 
+def _slot_sessions_manifest_path(campaign_id: str, slot: str) -> Path:
+    """Path to a slot's SESSION-LOG manifest sidecar (F08-1).
+
+    Lives under ``slots/.manifests/<slot>.json`` — a dedicated subdir so the non-recursive
+    ``slots/*.json`` glob in :func:`list_slots` never mistakes it for a restore point. The
+    manifest records each session file's byte length at slot time, so :func:`load_slot` can
+    roll the session logs back to match the rolled-back snapshot (a slot that restored only
+    snapshot.json left a discarded timeline — an undone TPK, a post-slot orphan session —
+    permanently canon in read_log_all / recap, the DM's lean-beat memory)."""
+    return _slots_dir(campaign_id) / ".manifests" / f"{safe_path_segment(slot, 'slot')}.json"
+
+
+def _session_files(campaign_id: str) -> dict[str, Path]:
+    """The campaign's live session-log files: ``{filename: path}`` for ``sessions/*.jsonl``
+    (non-recursive — the same glob shape read_log_all sees, so archived files under a
+    ``rolled-back-*`` subdir are invisible here too)."""
+    sessions_dir = _campaign_dir(campaign_id) / "sessions"
+    if not sessions_dir.is_dir():
+        return {}
+    return {p.name: p for p in sessions_dir.glob("*.jsonl") if p.is_file()}
+
+
 def save_slot(campaign_id: str, slot: str = "quicksave") -> Path:
     """Copy a campaign's CURRENT live snapshot into a named save slot.
 
     A slot is a point-in-time copy of the whole campaign aggregate, written atomically beside
     the live snapshot (campaigns/<id>/slots/<slot>.json). The live snapshot.json is the unit of
     persistence the engine already maintains, so we copy IT verbatim (not a re-serialized model)
     — the slot is byte-for-byte the campaign as last saved. Raises ValueError if the campaign has
-    no live snapshot yet. Caller holds campaign_lock (sole-writer)."""
+    no live snapshot yet. Caller holds campaign_lock (sole-writer).
+
+    F08-1: ALSO captures a SESSION-LOG manifest (each ``sessions/*.jsonl`` file's byte length at
+    slot time) into a sidecar, so a later load_slot can roll the append-only logs back to match
+    the rolled-back snapshot instead of leaving a discarded timeline canon in recap/lean-memory."""
     live = _campaign_dir(campaign_id) / "snapshot.json"
     if not live.exists():
         raise ValueError(f"no live snapshot for campaign {campaign_id!r} to save")
     data = live.read_text(encoding="utf-8")
     dest = _slot_path(campaign_id, slot)
     _atomic_write(dest, data)
+    # Capture the session-log state alongside the snapshot. The manifest maps each live session
+    # file name -> its current byte length; load_slot uses it to archive post-slot orphans and
+    # truncate grown sessions back to here. Best-effort: a manifest-write failure must not abort
+    # the slot save (the slot still restores the snapshot; logs degrade to today's behavior).
+    import json
+    try:
+        manifest = {name: p.stat().st_size for name, p in _session_files(campaign_id).items()}
+        _atomic_write(_slot_sessions_manifest_path(campaign_id, slot), json.dumps(manifest, indent=2))
+    except OSError as exc:
+        log.warning("save_slot(%s,%s): could not write sessions manifest: %s", campaign_id, slot, exc)
     return dest
 
 
+def _rollback_sessions_to_manifest(campaign_id: str, slot: str) -> None:
+    """Roll the campaign's session logs back to the state captured in a slot's manifest (F08-1).
+
+    Called by :func:`load_slot` AFTER the snapshot is restored, under the caller's campaign_lock
+    (sole-writer). For every live ``sessions/*.jsonl`` file:
+      * NOT in the manifest (created after the slot) -> ARCHIVE the whole file (a post-slot
+        orphan timeline) under ``sessions/rolled-back-<ts>/``;
+      * LONGER than its manifest byte length (grown after the slot) -> archive the discarded
+        TAIL, then truncate the live file back to the manifested length;
+      * SHORTER than or equal to the manifest length -> LEAVE AS-IS (degrade, never pad/raise:
+        a shorter file can arise after an intervening restore of an older slot).
+    Archiving (not deleting) keeps the discarded timeline recoverable; the archive subdir is
+    invisible to read_log_all's non-recursive ``*.jsonl`` glob, so recap/lean-memory match the
+    rolled-back snapshot. A MANIFEST-LESS slot (legacy, pre-F08-1) is a no-op -> today's
+    behavior (snapshot restored, logs untouched)."""
+    import json
+    mpath = _slot_sessions_manifest_path(campaign_id, slot)
+    if not mpath.exists():
+        return  # legacy / manifest-less slot: degrade to today (leave session logs alone)
+    try:
+        manifest: dict = json.loads(mpath.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        # A corrupt manifest must not brick a restore — degrade to today (logs untouched).
+        log.warning("load_slot(%s,%s): unreadable sessions manifest, leaving logs as-is: %s",
+                    campaign_id, slot, exc)
+        return
+
+    live = _session_files(campaign_id)
+    if not live:
+        return
+    archive_dir = _campaign_dir(campaign_id) / "sessions" / f"rolled-back-{int(time.time() * 1000)}"
+    for name, path in live.items():
+        try:
+            if name not in manifest:
+                # A whole session created after the slot — archive it out of the live set.
+                archive_dir.mkdir(parents=True, exist_ok=True)
+                os.replace(path, archive_dir / name)
+                continue
+            kept_len = int(manifest[name])
+            cur_len = path.stat().st_size
+            if cur_len <= kept_len:
+                continue  # unchanged or shorter — leave as-is (never pad/raise)
+            # Grown: archive the discarded tail, then truncate the live file back to slot length.
+            with open(path, "rb") as f:
+                head = f.read(kept_len)
+                tail = f.read()
+            archive_dir.mkdir(parents=True, exist_ok=True)
+            (archive_dir / name).write_bytes(tail)
+            # Atomic truncate via temp-then-replace so a crash never leaves a half-written log.
+            tmp = path.with_suffix(path.suffix + ".tmp")
+            with open(tmp, "wb") as f:
+                f.write(head)
+                f.flush()
+                os.fsync(f.fileno())
+            os.replace(tmp, path)
+        except OSError as exc:
+            # One unreadable/locked session file must not abort the whole rollback.
+            log.warning("load_slot(%s,%s): could not roll back session %s: %s",
+                        campaign_id, slot, name, exc)
+
+
 def load_slot(campaign_id: str, slot: str = "quicksave") -> Campaign:
     """Restore a named save slot back over the live campaign snapshot.
 
     Reads the slot, validates it parses as a Campaign for THIS campaign id (a slot belongs to the
     campaign it was saved from — we refuse to clobber the live state with a foreign/corrupt
     snapshot), then writes it to live via save_campaign (atomic + version-stamped). Raises
     FileNotFoundError if the slot is absent and ValueError if it is corrupt or mismatched.
-    OVERWRITES the live snapshot — caller holds campaign_lock and has confirmed intent."""
+    OVERWRITES the live snapshot — caller holds campaign_lock and has confirmed intent.
+
+    F08-1: ALSO rolls the append-only SESSION LOGS back to the slot's manifest, so a discarded
+    timeline (an undone TPK, a post-slot orphan session) can't stay canon in read_log_all /
+    recap (the DM's lean-beat memory). A manifest-less (legacy) slot leaves logs untouched."""
     src = _slot_path(campaign_id, slot)
     if not src.exists():
         raise FileNotFoundError(f"no save slot {slot!r} for campaign {campaign_id!r}")
@@ -220,6 +321,8 @@ def load_slot(campaign_id: str, slot: str = "quicksave") -> Campaign:
             f"save slot {slot!r} belongs to campaign {c.id!r}, not {campaign_id!r}; refusing to restore"
         )
     save_campaign(c)  # atomic replace of the live snapshot.json + fresh version stamp
+    # Fence the session logs to the slot AFTER the snapshot lands (both under the caller's lock).
+    _rollback_sessions_to_manifest(campaign_id, slot)
     return c