Merge pull request #22 from nickroci/fix/scholar-counter-preservation

fix(daemon): preserve server-owned counters on rewrite; wire fired co…

Author: nickroci

daemon/agent_mem_daemon/decay.py

@@ -171,6 +171,53 @@ def stamp_last_surfaced(
     return _atomic_write_entry(entry_path, fm, body)
 
 
+def _fired_count(fm: Dict[str, Any]) -> int:
+    """Current ``fired`` counter as a non-negative int (0 if absent or
+    unparseable — same permissive coercion as ``reinforced``)."""
+    raw = fm.get("fired")
+    if raw is None:
+        return 0
+    try:
+        n = int(raw)
+    except (TypeError, ValueError):
+        return 0
+    return n if n > 0 else 0
+
+
+def record_surface(
+    entry_path: Path,
+    *,
+    today_iso: Optional[str] = None,
+) -> bool:
+    """Record one priming surface event on the entry: increment ``fired``
+    by 1 AND stamp ``last_surfaced`` to today, in a single atomic write.
+
+    Unlike :func:`stamp_last_surfaced`, ``fired`` is NOT daily-gated — it
+    counts every surface event so the invariant ``fired-helpful <= fired``
+    holds (``fired-helpful`` increments per (session, entry, turn), which
+    can be several within a single day). ``last_surfaced`` is still set to
+    today either way; the date is naturally idempotent so re-stamping it is
+    harmless.
+
+    Args:
+        entry_path: absolute path to the entry's ``.md`` file.
+        today_iso: ISO date string to write (default: UTC today).
+            Tests pin this for deterministic assertions.
+
+    Returns:
+        True if the file was updated, False on any failure path (file
+        missing, bad frontmatter, write failed). Never raises.
+    """
+    iso = today_iso or _today_iso()
+    parsed = _read_and_parse_frontmatter(entry_path)
+    if parsed is None:
+        return False
+    fm, _raw_fm, body = parsed
+    fm["fired"] = _fired_count(fm) + 1
+    fm["last_surfaced"] = iso
+    return _atomic_write_entry(entry_path, fm, body)
+
+
 # ── Sweep state I/O ──────────────────────────────────────────────────
 
 
@@ -521,6 +568,7 @@ def maybe_run_sweep(
     "SweepResult",
     "maybe_run_sweep",
     "read_last_sweep_at",
+    "record_surface",
     "run_sweep",
     "should_sweep",
     "stamp_last_surfaced",

daemon/agent_mem_daemon/priming_rpc.py

@@ -294,10 +294,13 @@ def _post_render_bookkeeping(
 
     1. Per-session sent-cache update (only when body non-empty and
        session_id present — otherwise there's no state to track).
-    2. ``last_surfaced`` frontmatter stamp on each newly-shown entry,
-       so the decay sweep treats surfacing as activity. Runs even
-       without ``session_id`` because the agent saw the entries
-       regardless of how the daemon tracks sessions.
+    2. Surface bookkeeping on each newly-shown entry via
+       ``decay.record_surface``: increment the ``fired`` counter and
+       stamp ``last_surfaced``, so the decay sweep treats surfacing as
+       activity and ``fired`` tracks the surface denominator for
+       ``fired-helpful``. Runs even without ``session_id`` because the
+       agent saw the entries regardless of how the daemon tracks
+       sessions.
     3. Opportunistic ``maybe_run_sweep`` — self-skips unless the 24h
        cooldown has elapsed. One ``stat()`` on the sweep-state file
        on the common path.
@@ -310,9 +313,9 @@ def _post_render_bookkeeping(
     for link in newly_sent or ():
         entry_path = (kdir / f"{link}.md").resolve()
         try:
-            decay.stamp_last_surfaced(entry_path)
+            decay.record_surface(entry_path)
         except Exception:
-            log.exception("priming_rpc: last_surfaced stamp raised for %s", link)
+            log.exception("priming_rpc: surface bookkeeping raised for %s", link)
     try:
         decay.maybe_run_sweep(kdir)
     except Exception:

daemon/agent_mem_daemon/scholar_executor.py

@@ -90,6 +90,56 @@ def _reserialise(fm: Dict[str, Any], body: str) -> str:
     return f"---\n{dumped}\n---\n{body}"
 
 
+# Bookkeeping fields the daemon owns deterministically. The Scholar/Librarian
+# LLM is only *prompted* to preserve them and routinely emits the template
+# default (``0`` / absent), which clobbers the counters the daemon just bumped
+# on disk. So on any rewrite that REPLACES an existing entry, the executor is
+# the single source of truth: it forces each of these from the prior on-disk
+# entry onto the proposed frontmatter, discarding whatever the model emitted.
+# Everything else (title, keywords, status, confidence, applies-when, sources,
+# prose, …) stays LLM-owned and untouched.
+_SERVER_OWNED_FIELDS = (
+    "fired",
+    "fired-helpful",
+    "last_fired_helpful",
+    "reinforced",
+    "last_reinforced",
+    "last_surfaced",
+    "created",
+)
+
+
+def _preserve_server_owned_fields(abs_path: Path, proposed_body: str) -> str:
+    """Force the server-owned bookkeeping fields from the EXISTING on-disk
+    entry at ``abs_path`` onto ``proposed_body``'s frontmatter.
+
+    When ``abs_path`` has no pre-existing file (a brand-new entry) or its
+    frontmatter is unparseable, return ``proposed_body`` unchanged — there is
+    nothing to preserve. A server-owned field absent from the prior entry is
+    left as whatever the proposed body carries (defaults stand); a present one
+    overrides the LLM value verbatim.
+    """
+    if not abs_path.exists():
+        return proposed_body
+    prior_fm, _prior_body = _split_body(_safe_read(abs_path))
+    if not prior_fm:
+        return proposed_body
+    proposed_fm, body = _split_body(proposed_body)
+    if not proposed_fm:
+        # No frontmatter to graft onto — write the body as-is rather than
+        # synthesising a frontmatter block the model didn't intend.
+        return proposed_body
+    changed = False
+    for key in _SERVER_OWNED_FIELDS:
+        if key in prior_fm:
+            if proposed_fm.get(key) != prior_fm[key]:
+                proposed_fm[key] = prior_fm[key]
+                changed = True
+    if not changed:
+        return proposed_body
+    return _reserialise(proposed_fm, body)
+
+
 def _atomic_write(path: Path, text: str) -> None:
     """Write ``text`` to ``path`` atomically (tmp + os.replace)."""
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -257,9 +307,13 @@ def _apply_write_like(
     now: datetime,
     result: ExecResult,
 ) -> None:
-    """Shared body for write_entry / update_entry (write the body verbatim,
-    sync the index row, append the log)."""
+    """Shared body for write_entry / update_entry. When the entry already
+    exists on disk (update, or a write that replaces an entry), the
+    server-owned bookkeeping counters are forced from the prior entry before
+    the verbatim write so an LLM-emitted default can't clobber them; then
+    sync the index row and append the log."""
     abs_path = (knowledge_dir / rel_path).resolve()
+    body = _preserve_server_owned_fields(abs_path, body)
     _atomic_write(abs_path, body)
     fm, fm_body = _split_body(body)
     row = _index_row(rel_path, fm, fm_body, session_id=session_id)
@@ -301,8 +355,12 @@ def _do_merge(
     kd: Path, a: "ScholarMergeEntries", *, session_id: str, now: datetime, result: ExecResult
 ) -> None:
     """Write the merged target, then archive every distinct source path."""
-    _atomic_write((kd / a.target_path).resolve(), a.target_body)
-    fm, fm_body = _split_body(a.target_body)
+    target_abs = (kd / a.target_path).resolve()
+    # If the merge overwrites an existing entry in place, keep its
+    # server-owned counters rather than the model's emitted defaults.
+    target_body = _preserve_server_owned_fields(target_abs, a.target_body)
+    _atomic_write(target_abs, target_body)
+    fm, fm_body = _split_body(target_body)
     _upsert_index_row(
         kd, a.target_path, _index_row(a.target_path, fm, fm_body, session_id=session_id)
     )
@@ -418,8 +476,11 @@ def _do_abstract(
     read is recorded as a per-child miss but does not abort the action; the
     parent and the readable backlinks are still applied (integrity-first)."""
     parent_abs = (kd / a.parent_path).resolve()
-    _atomic_write(parent_abs, a.parent_body)
-    fm, fm_body = _split_body(a.parent_body)
+    # Abstraction usually writes a fresh parent, but if it overwrites an
+    # existing entry, keep that entry's server-owned counters.
+    parent_body = _preserve_server_owned_fields(parent_abs, a.parent_body)
+    _atomic_write(parent_abs, parent_body)
+    fm, fm_body = _split_body(parent_body)
     _upsert_index_row(
         kd, a.parent_path, _index_row(a.parent_path, fm, fm_body, session_id=session_id)
     )

daemon/tests/test_decay.py

@@ -42,6 +42,7 @@ def _write_entry(
     created: str = "2026-01-01",
     updated: str | None = None,
     status: str | None = None,
+    fired: int | None = None,
 ) -> None:
     """Create an entry with the supplied frontmatter fields."""
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -51,6 +52,8 @@ def _write_entry(
         "created": created,
         "reinforced": reinforced,
     }
+    if fired is not None:
+        fm["fired"] = fired
     if updated is not None:
         fm["updated"] = updated
     if last_surfaced is not None:
@@ -128,6 +131,97 @@ def test_stamp_last_surfaced_returns_false_for_no_frontmatter(home: Path) -> Non
     assert decay.stamp_last_surfaced(p, today_iso="2026-05-27") is False
 
 
+# ── record_surface (fired counter; Defect 2) ─────────────────────────
+
+
+def test_record_surface_increments_fired_and_stamps_last_surfaced(home: Path) -> None:
+    p = home / "knowledge" / "global" / "foo.md"
+    _write_entry(p, id_="foo", title="Foo", fired=0)
+    ok = decay.record_surface(p, today_iso="2026-05-27")
+    assert ok is True
+    fm = yaml.safe_load(p.read_text(encoding="utf-8").split("---")[1])
+    assert fm["fired"] == 1
+    assert str(fm["last_surfaced"]) == "2026-05-27"
+
+
+def test_record_surface_is_per_event_not_daily_gated(home: Path) -> None:
+    """The crux of the invariant: two surface events on the SAME day must
+    bump ``fired`` twice (unlike ``last_surfaced``, which is daily-idempotent)
+    so ``fired-helpful`` — which can bump several times a day — never exceeds
+    ``fired``."""
+    p = home / "knowledge" / "global" / "foo.md"
+    _write_entry(p, id_="foo", title="Foo", fired=0)
+    decay.record_surface(p, today_iso="2026-05-27")
+    decay.record_surface(p, today_iso="2026-05-27")
+    fm = yaml.safe_load(p.read_text(encoding="utf-8").split("---")[1])
+    assert fm["fired"] == 2
+    assert str(fm["last_surfaced"]) == "2026-05-27"
+
+
+def test_record_surface_starts_from_zero_when_field_absent(home: Path) -> None:
+    """An entry with no ``fired`` field starts the counter at 1, not crash."""
+    p = home / "knowledge" / "global" / "foo.md"
+    _write_entry(p, id_="foo", title="Foo")  # no fired field written
+    assert "fired:" not in p.read_text(encoding="utf-8")
+    decay.record_surface(p, today_iso="2026-05-27")
+    fm = yaml.safe_load(p.read_text(encoding="utf-8").split("---")[1])
+    assert fm["fired"] == 1
+
+
+def test_record_surface_coerces_bad_fired_value(home: Path) -> None:
+    """A non-int ``fired`` (corrupt frontmatter) is treated as 0, then +1."""
+    p = home / "knowledge" / "global" / "foo.md"
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_text(
+        "---\nid: foo\ntitle: Foo\ncreated: '2026-01-01'\nfired: not-a-number\n---\n\nbody\n"
+    )
+    decay.record_surface(p, today_iso="2026-05-27")
+    fm = yaml.safe_load(p.read_text(encoding="utf-8").split("---")[1])
+    assert fm["fired"] == 1
+
+
+def test_record_surface_preserves_other_frontmatter(home: Path) -> None:
+    p = home / "knowledge" / "global" / "foo.md"
+    _write_entry(p, id_="foo", title="Foo", fired=4, reinforced=3, last_reinforced="2026-04-01")
+    decay.record_surface(p, today_iso="2026-05-27")
+    fm = yaml.safe_load(p.read_text(encoding="utf-8").split("---")[1])
+    assert fm["id"] == "foo"
+    assert fm["reinforced"] == 3
+    assert fm["last_reinforced"] == "2026-04-01"
+    assert fm["fired"] == 5
+
+
+def test_record_surface_returns_false_for_missing_file(home: Path) -> None:
+    p = home / "knowledge" / "global" / "nonexistent.md"
+    assert decay.record_surface(p, today_iso="2026-05-27") is False
+
+
+def test_record_surface_returns_false_for_no_frontmatter(home: Path) -> None:
+    p = home / "knowledge" / "global" / "raw.md"
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_text("# Just a heading\n\nNo frontmatter here.\n")
+    assert decay.record_surface(p, today_iso="2026-05-27") is False
+
+
+def test_fired_helpful_never_exceeds_fired_under_surface_semantics(home: Path) -> None:
+    """Invariant guard: with N surface events and M<=N helpful bumps in the
+    same day, ``fired-helpful <= fired`` holds. ``record_surface`` counts each
+    surface; ``fired-helpful`` (modelled here as a same-day multi-bump) stays
+    bounded by it."""
+    p = home / "knowledge" / "global" / "foo.md"
+    _write_entry(p, id_="foo", title="Foo", fired=0)
+    # Three surface events today.
+    for _ in range(3):
+        decay.record_surface(p, today_iso="2026-05-27")
+    fm = yaml.safe_load(p.read_text(encoding="utf-8").split("---")[1])
+    fired = int(fm["fired"])
+    # The Scholar would bump fired-helpful at most once per surfaced turn; the
+    # denominator (fired) must dominate. Two helpful bumps <= three surfaces.
+    fired_helpful = 2
+    assert fired == 3
+    assert fired_helpful <= fired
+
+
 # ── sweep-state I/O ──────────────────────────────────────────────────
 
 

daemon/tests/test_priming_rpc.py

@@ -808,13 +808,14 @@ def test_post_render_bookkeeping_stamps_last_surfaced(
     home_with_isolated_paths, monkeypatch
 ) -> None:
     """Each entry in ``newly_sent`` should get its frontmatter
-    ``last_surfaced`` updated."""
+    ``last_surfaced`` stamped and its ``fired`` counter bumped."""
     home = home_with_isolated_paths
     k = home / "knowledge"
     entry = k / "global" / "foo.md"
     entry.parent.mkdir(parents=True, exist_ok=True)
     entry.write_text(
-        "---\nid: foo\ntitle: Foo\ncreated: '2026-01-01'\nreinforced: 0\n---\n\n# Foo\n\nbody.\n"
+        "---\nid: foo\ntitle: Foo\ncreated: '2026-01-01'\n"
+        "fired: 0\nreinforced: 0\n---\n\n# Foo\n\nbody.\n"
     )
     # Skip the sweep — separate concern, separately tested.
     monkeypatch.setattr(
@@ -825,6 +826,7 @@ def test_post_render_bookkeeping_stamps_last_surfaced(
     priming_rpc._post_render_bookkeeping(k, "test-session", "rendered body", ["global/foo"])
     text = entry.read_text(encoding="utf-8")
     assert "last_surfaced" in text
+    assert "fired: 1" in text
 
 
 def test_post_render_bookkeeping_records_session_cache(
@@ -843,7 +845,7 @@ def test_post_render_bookkeeping_records_session_cache(
     )
     monkeypatch.setattr(
         priming_rpc.decay,
-        "stamp_last_surfaced",
+        "record_surface",
         lambda *args, **kwargs: True,
     )
     priming_rpc._post_render_bookkeeping(
@@ -865,18 +867,18 @@ def test_post_render_bookkeeping_no_session_skips_cache_record(
         "maybe_run_sweep",
         lambda *args, **kwargs: None,
     )
-    stamped: list[str] = []
+    surfaced: list[str] = []
     monkeypatch.setattr(
         priming_rpc.decay,
-        "stamp_last_surfaced",
-        lambda path, **kwargs: stamped.append(str(path)) or True,
+        "record_surface",
+        lambda path, **kwargs: surfaced.append(str(path)) or True,
     )
     priming_rpc._post_render_bookkeeping(k, None, "rendered", ["global/foo"])
     # Cache untouched (empty).
     with priming_rpc._sent_cache_lock:
         assert len(priming_rpc._sent_cache) == 0
-    # But stamp still ran.
-    assert len(stamped) == 1
+    # But surface bookkeeping still ran.
+    assert len(surfaced) == 1
 
 
 def test_post_render_bookkeeping_swallows_stamp_failure(
@@ -889,7 +891,7 @@ def test_post_render_bookkeeping_swallows_stamp_failure(
     def _boom(*args, **kwargs):
         raise RuntimeError("disk full")
 
-    monkeypatch.setattr(priming_rpc.decay, "stamp_last_surfaced", _boom)
+    monkeypatch.setattr(priming_rpc.decay, "record_surface", _boom)
     monkeypatch.setattr(priming_rpc.decay, "maybe_run_sweep", lambda *args, **kwargs: None)
     # Should not raise.
     priming_rpc._post_render_bookkeeping(k, "s", "body", ["global/foo"])
@@ -906,7 +908,7 @@ def _boom(*args, **kwargs):
         raise RuntimeError("filesystem error")
 
     monkeypatch.setattr(priming_rpc.decay, "maybe_run_sweep", _boom)
-    monkeypatch.setattr(priming_rpc.decay, "stamp_last_surfaced", lambda *args, **kwargs: True)
+    monkeypatch.setattr(priming_rpc.decay, "record_surface", lambda *args, **kwargs: True)
     priming_rpc._post_render_bookkeeping(k, "s", "body", ["global/foo"])