feat(rebuild_log): instrument UserPromptSubmit path (#288 phase-1a)

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
 

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:<surviving_belief_id>``. 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 --------------------------------------------------------
 
 

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,