diff --git a/docs/rebuild_eval_harness.md b/docs/rebuild_eval_harness.md index 4285bda2f..dc8db9e9e 100644 --- a/docs/rebuild_eval_harness.md +++ b/docs/rebuild_eval_harness.md @@ -136,12 +136,29 @@ here) can reap files older than 30 days. fail-soft contract as `_write_telemetry` in `hook.py:358` — any I/O error is logged to stderr and never breaks the rebuild. -**Where the write hook lives.** Inside `rebuild()` in -`context_rebuilder.py`, immediately after the candidate list is -finalized but before the rendered block is returned. Writing from -the rebuilder (not the hook caller) means the log captures every -rebuild call site, including the PreCompact path and any future -direct caller, not only `UserPromptSubmit`. +**Where the write hook lives.** Two call sites: + +1. Inside `rebuild_v14()` in `context_rebuilder.py`, immediately + after the candidate list is finalized but before the rendered + block is returned. Catches the PreCompact path and any future + direct `rebuild_v14` caller. +2. Inside `user_prompt_submit()` in `hook.py`, after content-hash + dedup, via the `record_user_prompt_submit_log` helper in + `context_rebuilder.py`. Catches the high-frequency UPS retrieval + path, which calls `search_for_prompt` directly and never reaches + `rebuild_v14`. + +Both call sites share the schema, the `_append_rebuild_log_record` +writer, the size cap, and the env / TOML opt-out. UPS records carry +a synthetic single-turn `RecentTurn` derived from the prompt; the +on-disk record shape is identical to the PreCompact one. + +The original spec assumed all rebuild call sites went through +`rebuild_v14`, so phase-1a wired only that path. The UPS path +bypasses `rebuild_v14` entirely (`search_for_prompt` → +`retrieve()` returns the final hit list, no rebuild block built), +so a phase-1a-only ship produced empty logs under normal session +load — phase-1b was unreachable until UPS was wired. ### Layer 2: fixed-corpus precision harness diff --git a/src/aelfrice/context_rebuilder.py b/src/aelfrice/context_rebuilder.py index 7ee4f8141..f74340432 100644 --- a/src/aelfrice/context_rebuilder.py +++ b/src/aelfrice/context_rebuilder.py @@ -954,6 +954,105 @@ def _append_rebuild_log_record( ) +def record_user_prompt_submit_log( + *, + prompt: str, + session_id: str | None, + hits_pre_dedup: list[Belief], + hits_post_dedup: list[Belief], + log_path: Path | None, + enabled: bool = DEFAULT_REBUILD_LOG_ENABLED, + stderr: IO[str] | None = None, +) -> None: + """Emit one rebuild_log row for a UserPromptSubmit retrieval. + + Phase-1a wired the per-rebuild log only into ``rebuild_v14`` — + fired by ``PreCompact`` and rare. The high-frequency retrieval + path is ``user_prompt_submit``, which calls + ``hook_search.search_for_prompt`` directly. Without this hook, + an operator-week of normal use produces no rebuild_log rows and + phase-1b cannot accumulate data. + + Schema is the same Layer-1 record the spec ratifies in + ``docs/rebuild_eval_harness.md``: synthesise a single + ``RecentTurn`` from the prompt so the existing + ``_build_rebuild_log_record`` machinery (hash, extracted_query, + extracted_entities) applies unchanged. Candidates are the + pre-dedup hit list; pre-dedup hits that survive content-hash + dedup are ``packed``, the rest are ``dropped`` with reason + ``content_hash_collision_with:``. Score + fields are ``None`` per ``_empty_scores`` — the BM25 / posterior + decomposition is not exposed at this call site, and locking the + schema in phase-1a means phase-2 ranker work fills the same + fields without a log-format migration. + + No-op when ``enabled`` is False, when the env opt-out is set, or + when ``log_path`` is None / ``hits_pre_dedup`` is empty (mirrors + ``rebuild_v14``: no candidate set, no row). + """ + if not enabled: + return + if _rebuild_log_disabled_via_env(): + return + if log_path is None: + return + if not hits_pre_dedup: + return + surviving_ids: set[str] = {b.id for b in hits_post_dedup} + survivor_by_hash: dict[str, str] = {} + for b in hits_post_dedup: + survivor_by_hash.setdefault( + hashlib.sha1(b.content.encode("utf-8")).hexdigest(), b.id, + ) + candidates: list[dict[str, object]] = [] + n_dropped_by_dedup = 0 + for rank, b in enumerate(hits_pre_dedup, start=1): + if b.id in surviving_ids: + decision = "packed" + reason: str | None = None + else: + decision = "dropped" + digest = hashlib.sha1(b.content.encode("utf-8")).hexdigest() + survivor = survivor_by_hash.get(digest) + reason = ( + f"content_hash_collision_with:{survivor}" + if survivor + else "content_hash_collision" + ) + n_dropped_by_dedup += 1 + candidates.append({ + "belief_id": b.id, + "rank": rank, + "scores": _empty_scores(), + "lock_level": _belief_lock_level_for_log(b), + "decision": decision, + "reason": reason, + }) + pack_summary: dict[str, int] = { + "n_candidates": len(hits_pre_dedup), + "n_packed": len(hits_post_dedup), + # The UPS path has no visibility into floor / budget drops: + # ranking happens inside `retrieve()` and only the surviving + # set crosses the function boundary. Holding these at zero + # keeps the on-disk schema stable; phase-2 wiring will fill + # them when the ranker exposes its drop reasons. + "n_dropped_by_floor": 0, + "n_dropped_by_dedup": n_dropped_by_dedup, + "n_dropped_by_budget": 0, + "total_chars_packed": sum(len(b.content) for b in hits_post_dedup), + } + synthetic_turn = RecentTurn( + role="user", text=prompt, session_id=session_id, + ) + record = _build_rebuild_log_record( + recent_turns=[synthetic_turn], + session_id=session_id, + candidates=candidates, + pack_summary=pack_summary, + ) + _append_rebuild_log_record(log_path, record, stderr=stderr) + + # --- Format helpers -------------------------------------------------------- diff --git a/src/aelfrice/hook.py b/src/aelfrice/hook.py index 4d95807ff..ba694ee83 100644 --- a/src/aelfrice/hook.py +++ b/src/aelfrice/hook.py @@ -665,9 +665,21 @@ def user_prompt_submit( n_unique = len(unique_hashes) n_l0 = sum(1 for h in hits if h.lock_level == LOCK_USER) n_l1 = n_returned - n_l0 + hits_pre_dedup = list(hits) # AC6: optional dedup before formatting. if config.collapse_duplicate_hashes: hits = _dedup_by_content_hash(hits) + # #288 phase-1a extension: emit one rebuild_log row per + # UPS retrieval. Without this the high-frequency rebuild + # call site produces no log; phase-1b operator-week data + # collection depends on it. + _emit_user_prompt_submit_rebuild_log( + prompt=prompt, + session_id=session_id, + hits_pre_dedup=hits_pre_dedup, + hits_post_dedup=hits, + stderr=serr, + ) # total_chars measured post-collapse (what is actually injected). total_chars = sum(len(h.content) for h in hits) body = _format_hits(hits) @@ -701,6 +713,51 @@ def user_prompt_submit( return 0 +def _emit_user_prompt_submit_rebuild_log( + *, + prompt: str, + session_id: str | None, + hits_pre_dedup: list[Belief], + hits_post_dedup: list[Belief], + stderr: IO[str] | None = None, +) -> None: + """Append a phase-1a rebuild_log row for this UPS retrieval. + + Fail-soft: any path-resolution or import failure traces one + line to stderr and never propagates. The rebuild_log is + diagnostic; a write error must not break the hook. + """ + serr = stderr if stderr is not None else sys.stderr + try: + from aelfrice.context_rebuilder import ( # noqa: PLC0415 + _rebuild_log_dir_for_db, + load_rebuilder_config, + record_user_prompt_submit_log, + ) + + if not session_id: + return + p = db_path() + if str(p) == ":memory:": + return + log_path = _rebuild_log_dir_for_db(p) / f"{session_id}.jsonl" + rebuilder_cfg = load_rebuilder_config() + record_user_prompt_submit_log( + prompt=prompt, + session_id=session_id, + hits_pre_dedup=hits_pre_dedup, + hits_post_dedup=hits_post_dedup, + log_path=log_path, + enabled=rebuilder_cfg.rebuild_log_enabled, + stderr=serr, + ) + except Exception as exc: + print( + f"aelfrice: UPS rebuild_log emit failed (non-fatal): {exc}", + file=serr, + ) + + def _write_telemetry( *, prompt: str, diff --git a/tests/test_rebuild_log_user_prompt_submit.py b/tests/test_rebuild_log_user_prompt_submit.py new file mode 100644 index 000000000..81cf96ef7 --- /dev/null +++ b/tests/test_rebuild_log_user_prompt_submit.py @@ -0,0 +1,309 @@ +"""Tests for the #288 phase-1a rebuild_log on the UserPromptSubmit path. + +Phase-1a originally instrumented only `rebuild_v14` (the PreCompact +call site). The UserPromptSubmit hook calls `search_for_prompt` +directly, so the high-frequency rebuild path produced no log rows +and phase-1b operator-week data could not accumulate. These tests +cover the UPS-side wiring: schema parity with the PreCompact log, +dedup-drop visibility, env opt-out, TOML opt-out, and fail-soft +behaviour when the log path can't be derived. +""" +from __future__ import annotations + +import io +import json +from pathlib import Path + +import pytest + +from aelfrice.context_rebuilder import ( + REBUILD_LOG_ENV, + record_user_prompt_submit_log, +) +from aelfrice.hook import user_prompt_submit +from aelfrice.models import BELIEF_FACTUAL, LOCK_NONE, LOCK_USER, Belief +from aelfrice.store import MemoryStore + + +# ---- helpers ----------------------------------------------------------- + + +def _mk( + bid: str, + content: str, + lock_level: str = LOCK_NONE, + locked_at: str | None = None, +) -> Belief: + return Belief( + id=bid, + content=content, + content_hash=f"h_{bid}", + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=lock_level, + locked_at=locked_at, + demotion_pressure=0, + created_at="2026-04-28T00:00:00Z", + last_retrieved_at=None, + ) + + +def _seed_db(db_path: Path, beliefs: list[Belief]) -> None: + store = MemoryStore(str(db_path)) + try: + for b in beliefs: + store.insert_belief(b) + finally: + store.close() + + +def _payload(prompt: str, session_id: str = "ups-sess-1") -> str: + return json.dumps( + { + "session_id": session_id, + "transcript_path": "/dev/null", + "cwd": "/tmp", + "hook_event_name": "UserPromptSubmit", + "prompt": prompt, + } + ) + + +def _read_log(log_path: Path) -> list[dict[str, object]]: + return [ + json.loads(ln) + for ln in log_path.read_text(encoding="utf-8").splitlines() + if ln.strip() + ] + + +# ---- end-to-end through the hook entry point --------------------------- + + +def test_user_prompt_submit_writes_rebuild_log( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "memory.db" + _seed_db(db, [_mk("F1", "the kitchen is full of bananas")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + + sout = io.StringIO() + rc = user_prompt_submit( + stdin=io.StringIO(_payload("bananas", "ups-sess-1")), + stdout=sout, + stderr=io.StringIO(), + ) + assert rc == 0 + assert sout.getvalue() != "" + + log_path = tmp_path / "rebuild_logs" / "ups-sess-1.jsonl" + assert log_path.exists(), ( + "UPS hook must emit a rebuild_log row when retrieval returns " + "any candidate" + ) + rows = _read_log(log_path) + assert len(rows) == 1 + rec = rows[0] + assert rec["session_id"] == "ups-sess-1" + assert isinstance(rec["ts"], str) and rec["ts"].endswith("Z") + # Schema parity with the PreCompact rebuild_log: same input/ + # candidates/pack_summary keys with the same `_empty_scores` + # block per candidate. + assert set(rec["input"]) == { + "recent_turns_hash", "n_recent_turns", + "extracted_query", "extracted_entities", "extracted_intent", + } + assert rec["input"]["n_recent_turns"] == 1 + assert isinstance(rec["candidates"], list) + assert len(rec["candidates"]) >= 1 + cand = rec["candidates"][0] + assert set(cand["scores"]) == { + "bm25", "posterior_mean", "reranker", "final", + } + assert cand["decision"] == "packed" + assert cand["reason"] is None + assert rec["pack_summary"]["n_candidates"] >= 1 + assert rec["pack_summary"]["n_packed"] >= 1 + + +def test_user_prompt_submit_no_log_when_no_hits( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """No retrieval hits means no candidate set, hence no row — + same contract as `rebuild_v14` on an empty store.""" + db = tmp_path / "memory.db" + _seed_db(db, [_mk("F1", "elephants are large")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + + rc = user_prompt_submit( + stdin=io.StringIO(_payload("dogs", "no-hit-sess")), + stdout=io.StringIO(), + stderr=io.StringIO(), + ) + assert rc == 0 + log_path = tmp_path / "rebuild_logs" / "no-hit-sess.jsonl" + assert not log_path.exists() + + +def test_user_prompt_submit_env_opt_out( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "memory.db" + _seed_db(db, [_mk("F1", "the kitchen is full of bananas")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + monkeypatch.setenv(REBUILD_LOG_ENV, "0") + + rc = user_prompt_submit( + stdin=io.StringIO(_payload("bananas", "opt-out")), + stdout=io.StringIO(), + stderr=io.StringIO(), + ) + assert rc == 0 + log_path = tmp_path / "rebuild_logs" / "opt-out.jsonl" + assert not log_path.exists() + + +def test_user_prompt_submit_toml_opt_out( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + db = tmp_path / "memory.db" + _seed_db(db, [_mk("F1", "the kitchen is full of bananas")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + cfg = tmp_path / ".aelfrice.toml" + cfg.write_text( + "[rebuild_log]\nenabled = false\n", + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + + rc = user_prompt_submit( + stdin=io.StringIO(_payload("bananas", "toml-off")), + stdout=io.StringIO(), + stderr=io.StringIO(), + ) + assert rc == 0 + log_path = tmp_path / "rebuild_logs" / "toml-off.jsonl" + assert not log_path.exists() + + +def test_user_prompt_submit_no_log_when_session_id_missing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """The on-disk JSONL is keyed by session_id; without it there's + no file path to write to. Drop silently rather than fabricating + a session id.""" + db = tmp_path / "memory.db" + _seed_db(db, [_mk("F1", "the kitchen is full of bananas")]) + monkeypatch.setenv("AELFRICE_DB", str(db)) + payload = json.dumps( + { + # no session_id field + "transcript_path": "/dev/null", + "cwd": "/tmp", + "hook_event_name": "UserPromptSubmit", + "prompt": "bananas", + } + ) + + rc = user_prompt_submit( + stdin=io.StringIO(payload), + stdout=io.StringIO(), + stderr=io.StringIO(), + ) + assert rc == 0 + log_dir = tmp_path / "rebuild_logs" + assert not log_dir.exists() or not any(log_dir.iterdir()) + + +# ---- direct unit coverage of the helper -------------------------------- + + +def test_record_user_prompt_submit_log_marks_dedup_drops( + tmp_path: Path, +) -> None: + """Pre-dedup hits with duplicate content collapse to one + `packed` survivor; the dropped duplicates carry + `content_hash_collision_with:` as their reason.""" + log_path = tmp_path / "rebuild_logs" / "dedup.jsonl" + pre = [ + _mk("A", "same content"), + _mk("B", "same content"), # collides with A + _mk("C", "different content"), + ] + post = [pre[0], pre[2]] # B dropped by dedup + record_user_prompt_submit_log( + prompt="anything", + session_id="dedup", + hits_pre_dedup=pre, + hits_post_dedup=post, + log_path=log_path, + enabled=True, + stderr=io.StringIO(), + ) + rows = _read_log(log_path) + assert len(rows) == 1 + rec = rows[0] + assert rec["pack_summary"]["n_candidates"] == 3 + assert rec["pack_summary"]["n_packed"] == 2 + assert rec["pack_summary"]["n_dropped_by_dedup"] == 1 + by_id = {c["belief_id"]: c for c in rec["candidates"]} + assert by_id["A"]["decision"] == "packed" + assert by_id["B"]["decision"] == "dropped" + assert by_id["B"]["reason"] == "content_hash_collision_with:A" + assert by_id["C"]["decision"] == "packed" + + +def test_record_user_prompt_submit_log_lock_level_passthrough( + tmp_path: Path, +) -> None: + log_path = tmp_path / "rebuild_logs" / "locks.jsonl" + pre = [ + _mk("L0", "x", lock_level=LOCK_USER, locked_at="2026-04-28T00:00:00Z"), + _mk("L1", "y"), + ] + record_user_prompt_submit_log( + prompt="q", + session_id="locks", + hits_pre_dedup=pre, + hits_post_dedup=pre, + log_path=log_path, + enabled=True, + stderr=io.StringIO(), + ) + rec = _read_log(log_path)[0] + by_id = {c["belief_id"]: c for c in rec["candidates"]} + assert by_id["L0"]["lock_level"] == "user" + assert by_id["L1"]["lock_level"] == "none" + + +def test_record_user_prompt_submit_log_disabled_writes_nothing( + tmp_path: Path, +) -> None: + log_path = tmp_path / "rebuild_logs" / "off.jsonl" + record_user_prompt_submit_log( + prompt="q", + session_id="off", + hits_pre_dedup=[_mk("A", "x")], + hits_post_dedup=[_mk("A", "x")], + log_path=log_path, + enabled=False, + stderr=io.StringIO(), + ) + assert not log_path.exists() + + +def test_record_user_prompt_submit_log_no_path_is_noop( + tmp_path: Path, +) -> None: + """`log_path=None` means we couldn't derive a per-session file + (e.g. the brain-graph DB is in-memory). Must be a silent no-op.""" + record_user_prompt_submit_log( + prompt="q", + session_id="x", + hits_pre_dedup=[_mk("A", "x")], + hits_post_dedup=[_mk("A", "x")], + log_path=None, + enabled=True, + stderr=io.StringIO(), + )