diff --git a/CHANGELOG.md b/CHANGELOG.md index a3cc234ac..93a38f0dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,8 +44,9 @@ installable release; see the roadmap in [README.md](README.md). ### Added -- **`aelf doctor --prune-dormant` interactively deletes idle per-project DBs** ([#594](https://github.com/robotrocketscience/aelfrice/issues/594)). Companion to the #589 legacy-schema nag: scans `~/.aelfrice/projects/*/memory.db` for files whose mtime is older than `--idle-days` (default 30) and lists them with row count + size + idle days. Default action is dry-run; `--apply` prompts `[y/N]` per DB and deletes only on `y`/`yes`. There is no `--yes` shortcut — every deletion is per-DB and explicit, locking the issue body's "never silent delete" constraint. Detection is schema-agnostic (legacy and modern DBs are both pruneable when idle), and empty DBs / unreadable schemas surface with `row_count=0` rather than being skipped — an idle empty DB is the cleanest prune target. The filesystem-orphan detection mode listed in the issue body (project root no longer exists) is deferred: `project_warm.py` hashes the git-common-dir to a 12-char slug and stores no reverse mapping, so rebuilding `id → original-path` is a separate piece of work. `--projects-dir PATH` overrides the scan root for tests. +- **Eval-harness host-agent replay path: `replay_post_fork` writes/joins per-run JSONL** ([#600](https://github.com/robotrocketscience/aelfrice/issues/600)). Commit-2 of #592, redesigned away from a direct vendor-SDK dependency. `replay_post_fork` gains an optional `run_dir: Path | None` parameter; threading via the new `--run-dir` argparse flag on `eval_harness.py` propagates it through `run_one`, `sweep_thresholds`, and `sweep_budgets`. With `run_dir` set, the harness writes one JSON row per eval_turn (`{turn_idx, rebuilt_block, user_turn, expected}`) to `/__t__b/replay_requests.jsonl` and reads `replay_responses.jsonl` (if present) on the next invocation, joining by `turn_idx`. Substring-match (`expected.lower() in actual.lower()`) short-circuits to `matched=True`; filled rows without substring match drop to `reason="needs_llm_judge"` for commit-3 of #592. Missing or empty `actual` rows hold `reason="pending_replay"`. Without `--run-dir` the legacy `needs_replay_client` placeholder mode stays default — sweeps continue to produce valid latency + token-cost numbers without dispatching child tasks. `benchmarks/context-rebuilder/README.md` documents the operator flow (sweep → host dispatches one child task per request row → write responses → re-run sweep). aelfrice still imports zero from any vendor SDK in this surface; the model invocation is the host's responsibility, mirroring the `/aelf:onboard` polymorphic split. Five new tests (`test_replay_pending_round_trip`, etc.) cover the request-emit, response-join, partial-coverage, and per-(case, config) subdir behaviour; existing 15 wiring tests pass unchanged. +- **`aelf doctor --prune-dormant` interactively deletes idle per-project DBs** ([#594](https://github.com/robotrocketscience/aelfrice/issues/594)). Companion to the #589 legacy-schema nag: scans `~/.aelfrice/projects/*/memory.db` for files whose mtime is older than `--idle-days` (default 30) and lists them with row count + size + idle days. Default action is dry-run; `--apply` prompts `[y/N]` per DB and deletes only on `y`/`yes`. There is no `--yes` shortcut — every deletion is per-DB and explicit, locking the issue body's "never silent delete" constraint. Detection is schema-agnostic (legacy and modern DBs are both pruneable when idle), and empty DBs / unreadable schemas surface with `row_count=0` rather than being skipped — an idle empty DB is the cleanest prune target. The filesystem-orphan detection mode listed in the issue body (project root no longer exists) is deferred: `project_warm.py` hashes the git-common-dir to a 12-char slug and stores no reverse mapping, so rebuilding `id → original-path` is a separate piece of work. `--projects-dir PATH` overrides the scan root for tests. - **`aelf doctor` flags per-project DBs on pre-v1.x schema (no `origin` column)** ([#589](https://github.com/robotrocketscience/aelfrice/issues/589)). Scans all `~/.aelfrice/projects/*/memory.db` and appends a `legacy-schema per-project DBs detected` block listing each flagged DB with belief row count and idle days. DBs on the old schema cannot participate in the v2.x lifecycle (`agent_remembered`, `user_validated`, calibrated weights, `aelf:promote`). Block is quiet when no legacy DBs are found (parity with #557 quietness). `aelf migrate` is now visible in `--help` output (previously `argparse.SUPPRESS`) so the fix line in the nag block is discoverable. - **Context-rebuilder eval harness — opt-in LLM-judge stage for open-ended turns** ([#592](https://github.com/robotrocketscience/aelfrice/issues/592)). Commit-3 of #592. New `benchmarks/context-rebuilder/judges/llm_judge.py` adds a host-polymorphic judge that scores open-ended replay rows (those tagged `reason="needs_llm_judge"` by the deterministic substring path). `write_judge_requests(rows, run_dir, max_judge_calls=N)` writes `judge_requests.jsonl` carrying strictly `(turn_idx, expected, actual)` — the rebuilt block and user turn are deliberately excluded per `docs/BENCHMARKS.md` Pass 1 / Pass 2 separation; letting the judge see retrieval context would inflate fidelity. Default `max_judge_calls=0` disables the stage so CI runs are free and outbound model traffic is opt-in. The judge tier label is `"anchor"` — the operator maps this to their host CLI's calibration-baseline tier (the tier the prior Cohen's-κ disagreement methodology was measured at); a cheaper-tier knob is intentionally not exposed because cheaper judges weaken comparability on the open-ended turns this stage exists to score. Dispatch is operator-driven: the host CLI reads the request file, issues one off-band model call per row at the anchor tier with `JUDGE_PROMPT_TEMPLATE`, writes `judge_responses.jsonl`. `read_judge_responses` + `apply_judge_verdicts` then fold verdicts back into the replay rows (clearing `reason`, setting `matched`, adding `judge_rationale`). No provider SDK import; matches `/aelf:onboard`'s polymorphic-classify pattern. 15 deterministic tests in `tests/test_context_rebuilder_eval_judge.py` cover the contamination boundary, cost-cap binding, response round-trip, malformed-line skip, fold-verdicts purity, and the no-SDK guard. Harness-side wiring (so the stage rides on `eval_harness.py --run-dir` rather than the standalone helpers) joins onto PR #601's run-dir plumbing in a follow-up. diff --git a/benchmarks/context-rebuilder/README.md b/benchmarks/context-rebuilder/README.md index 6d1010d78..06ba07b09 100644 --- a/benchmarks/context-rebuilder/README.md +++ b/benchmarks/context-rebuilder/README.md @@ -277,6 +277,79 @@ These remain open for the fidelity scorer (#138): 4. Whether augment-mode loses fidelity vs. suppress-mode (matters for v2.x suppress-mode promotion decision). +## Host-agent eval-replay (#600) + +`replay_post_fork` participates in a polymorphic split that mirrors +the `/aelf:onboard` pattern: aelfrice never imports the +model-vendor SDK, never holds API keys, and pushes the model +invocation to the calling host (an MCP-enabled host CLI, an +operator-driven loop, or a private skill). + +### The flow + +1. **Sweep with `--run-dir`** writes per-(case, config) request + files to disk and skips the model call: + + ``` + uv run python benchmarks/context-rebuilder/eval_harness.py \ + --mode threshold-sweep \ + --corpus benchmarks/context-rebuilder/fixtures/synthetic/ \ + --out /tmp/sweep1.json \ + --run-dir /tmp/sweep1/ + ``` + + For each (case, threshold, budget) cell, the harness writes: + + /tmp/sweep1/__t__b/replay_requests.jsonl + + One JSON row per `eval_turn`: + + ```json + {"turn_idx": 8, "rebuilt_block": "...", "user_turn": "...", "expected": "..."} + ``` + + `user_turn` is the most-recent `role=user` text at-or-before the + eval index — the prompt the host will replay through the rebuilt + context. All rows in the harness JSON output carry + `reason=pending_replay`; `score_fidelity` returns 0 on this + pass. + +2. **Operator dispatches one child task per row.** From a host + session (an MCP-enabled host CLI, an operator-driven loop, or a + private replay-eval skill), iterate over each + `replay_requests.jsonl` line and dispatch a child task prompted + with `rebuilt_block + "\n---\n" + user_turn`. Capture the + reply as `actual` and append a JSON line to + `replay_responses.jsonl` in the same directory: + + ```json + {"turn_idx": 8, "actual": "..."} + ``` + + The dispatch step is operator-driven, not aelfrice. A private + `aelf:replay-eval` skill in `~/.claude/skills/` can automate the + for-loop; the contract is the on-disk request/response files. + +3. **Re-run the sweep**, same flags. `replay_post_fork` reads each + `replay_responses.jsonl`, joins by `turn_idx`, and: + + * If the response is missing or `actual` is empty: + `matched=False`, `reason=pending_replay` (unchanged from pass 1). + * If `expected.lower() in actual.lower()`: `matched=True`, + `reason=""` (substring half of the fidelity verdict). + * Otherwise: `matched=False`, `reason=needs_llm_judge` — + open-ended turns parked for commit-3 of #592 (LLM judge). + + `score_fidelity` reports the substring half of the verdict on + this pass. + +### Without `--run-dir` + +The legacy stub path stays default: every row holds +`reason=needs_replay_client`, `matched=False`, the `actual` field +is empty, and `score_fidelity` returns 0. Threshold/budget sweeps +still produce valid latency + token-cost numbers. + ## LLM-judge stage (commit-3 of #592) Open-ended replay rows that the deterministic substring scorer diff --git a/benchmarks/context-rebuilder/eval_harness.py b/benchmarks/context-rebuilder/eval_harness.py index d4a14ebf5..f4d47ff3f 100644 --- a/benchmarks/context-rebuilder/eval_harness.py +++ b/benchmarks/context-rebuilder/eval_harness.py @@ -214,70 +214,206 @@ def run_rebuilder( REPLAY_PENDING_REASON: Final[str] = "needs_replay_client" -"""Marker emitted by `replay_post_fork` until the model-invocation -client lands. Surfaced in `failures[].reason` so summary readers can -distinguish "skipped, judge required" from a real fidelity miss.""" +"""Marker emitted by `replay_post_fork` when no `run_dir` is given (the +v1.2.0 stub default). Surfaced in `failures[].reason` so summary readers +can distinguish "skipped, judge required" from a real fidelity miss.""" + +PENDING_REPLAY_REASON: Final[str] = "pending_replay" +"""Marker for a row whose request is now persisted to +`/replay_requests.jsonl` but whose response has not yet been +filled in. Distinct from `needs_replay_client`: pending_replay means +"requests written, awaiting host-agent dispatch + replay_responses.jsonl".""" + +NEEDS_LLM_JUDGE_REASON: Final[str] = "needs_llm_judge" +"""Marker for a row whose response is filled in but the substring/token +match fails. The open-ended fidelity verdict belongs to commit-3 of +#592 (LLM judge). Substring success short-circuits to matched=True.""" + +REPLAY_REQUESTS_FILENAME: Final[str] = "replay_requests.jsonl" +REPLAY_RESPONSES_FILENAME: Final[str] = "replay_responses.jsonl" + + +def _read_user_and_expected( + case: TranscriptCase, +) -> tuple[dict[int, str], dict[int, str]]: + """Walk `case.path` once. Return (expected_by_idx, user_turn_by_idx). + + `expected_by_idx[idx]` is the `text` at line `idx` (the eval target). + `user_turn_by_idx[idx]` is the most recent user-role `text` at-or- + before line `idx` — the prompt the harness will replay through the + rebuilt context to produce `actual`. + """ + expected_by_idx: dict[int, str] = {} + user_turn_by_idx: dict[int, str] = {} + if not case.eval_turns: + return expected_by_idx, user_turn_by_idx + wanted = set(case.eval_turns) + last_user_text = "" + try: + with case.path.open("r", encoding="utf-8") as f: + for i, raw in enumerate(f): + line = raw.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(obj, dict): + continue + text = obj.get("text") if isinstance(obj.get("text"), str) else "" + if obj.get("role") == "user": + last_user_text = text + if i in wanted: + expected_by_idx[i] = text + user_turn_by_idx[i] = last_user_text + except OSError: + return {}, {} + return expected_by_idx, user_turn_by_idx + + +def _read_replay_responses(run_dir: Path) -> dict[int, str]: + """Read `/replay_responses.jsonl`, one row per turn_idx. + + Each row is `{"turn_idx": int, "actual": str}`. Missing file → {}; + malformed lines are skipped silently. The host-agent dispatcher + (operator-driven, not aelfrice) is the writer; this function is + the read half of the polymorphic eval-replay split. + """ + out: dict[int, str] = {} + path = run_dir / REPLAY_RESPONSES_FILENAME + if not path.is_file(): + return out + try: + with path.open("r", encoding="utf-8") as f: + for raw in f: + line = raw.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(obj, dict): + continue + idx = obj.get("turn_idx") + actual = obj.get("actual") + if isinstance(idx, int) and isinstance(actual, str): + out[idx] = actual + except OSError: + return {} + return out + + +def _write_replay_requests( + run_dir: Path, + rebuilt_context: str, + eval_turns: tuple[int, ...], + expected_by_idx: dict[int, str], + user_turn_by_idx: dict[int, str], +) -> None: + """Write `/replay_requests.jsonl`. One row per eval_turn. + + Schema: `{turn_idx, rebuilt_block, user_turn, expected}`. Best-effort + — `mkdir` and write failures squash to a no-op so the rest of + `run_one` (latency, token cost) still completes. + """ + try: + run_dir.mkdir(parents=True, exist_ok=True) + path = run_dir / REPLAY_REQUESTS_FILENAME + with path.open("w", encoding="utf-8") as f: + for idx in eval_turns: + row = { + "turn_idx": idx, + "rebuilt_block": rebuilt_context, + "user_turn": user_turn_by_idx.get(idx, ""), + "expected": expected_by_idx.get(idx, ""), + } + f.write(json.dumps(row) + "\n") + except OSError: + return def replay_post_fork( rebuilt_context: str, case: TranscriptCase, + *, + run_dir: Path | None = None, ) -> list[dict]: """Replay turns fork_turn..end with rebuilt_context as session start. - **Stub until the model-invocation client lands** (#592 commit-2 / -3). - Returns one placeholder dict per `case.eval_turns`: - - {"turn_idx": int, "expected": str, "actual": "", - "matched": False, "reason": REPLAY_PENDING_REASON} + Two modes, selected by `run_dir`: + + **No `run_dir` (v1.2.0 stub default).** Returns one placeholder dict + per `case.eval_turns` with `actual=""`, `matched=False`, + `reason=REPLAY_PENDING_REASON`. Threshold/budget-sweep modes get + valid latency + token-cost numbers without dispatching child tasks. + + **With `run_dir` (host-agent eval-replay, #600).** Writes + `/replay_requests.jsonl` so an operator-driven dispatcher + (an MCP-enabled host CLI, an operator loop, or a private + `aelf:replay-eval` skill) can spawn one child task per row, + prompted with `rebuilt_block + "\\n---\\n" + user_turn`, expecting + the dispatcher to write `/replay_responses.jsonl`. On the next harness invocation + this function reads that response file, joins by `turn_idx`, and + fills `actual`: + + * `actual` empty / row missing → matched=False, reason=PENDING_REPLAY_REASON. + * `expected.lower() in actual.lower()` → matched=True (substring half + of the fidelity verdict). + * `actual` filled but no substring match → matched=False, + reason=NEEDS_LLM_JUDGE_REASON (open-ended; commit-3 of #592 LLM + judge picks this up). `expected` is pulled from the captured transcript's `text` field at - the eval-turn line index (1-indexed via the file order, matching the - fork_turn convention). When the line is missing or malformed the - expected text falls back to "" so the harness still produces a - coherent record. - - Threading this stub keeps `run_one` end-to-end runnable: `score_fidelity` - reads `matched=False` and returns 0.0; `failures` populates with a - machine-readable `reason` so threshold-sweep / budget-sweep modes - surface latency + token-cost numbers without crashing on the - fidelity step. The real model client lands in a follow-up commit. + the eval-turn line index (matching the fork_turn convention). The + aelfrice repo never imports a vendor model SDK on this path; the + model call is the host's responsibility. """ - expected_by_idx: dict[int, str] = {} - if case.eval_turns: - wanted = set(case.eval_turns) - try: - with case.path.open("r", encoding="utf-8") as f: - for i, line in enumerate(f): - if i not in wanted: - continue - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue - if isinstance(obj, dict): - t = obj.get("text") - if isinstance(t, str): - expected_by_idx[i] = t - except OSError: - # Unreadable file → return an empty list rather than partial - # placeholders. The harness's load_corpus already filtered - # this case out for the corpus-walking path; this guard is - # for direct-call use. - return [] - return [ - { - "turn_idx": idx, - "expected": expected_by_idx.get(idx, ""), - "actual": "", - "matched": False, - "reason": REPLAY_PENDING_REASON, - } - for idx in case.eval_turns - ] + expected_by_idx, user_turn_by_idx = _read_user_and_expected(case) + if case.eval_turns and not expected_by_idx and not user_turn_by_idx: + # _read_user_and_expected returned ({}, {}) only on OSError. + # Mirror the historical behaviour for unreadable files. + return [] + + responses_by_idx: dict[int, str] = {} + if run_dir is not None: + if case.eval_turns: + _write_replay_requests( + run_dir, + rebuilt_context, + case.eval_turns, + expected_by_idx, + user_turn_by_idx, + ) + responses_by_idx = _read_replay_responses(run_dir) + + out: list[dict] = [] + for idx in case.eval_turns: + expected = expected_by_idx.get(idx, "") + actual = responses_by_idx.get(idx, "") + if run_dir is None: + reason = REPLAY_PENDING_REASON + matched = False + elif not actual: + reason = PENDING_REPLAY_REASON + matched = False + elif expected and expected.lower() in actual.lower(): + reason = "" + matched = True + else: + reason = NEEDS_LLM_JUDGE_REASON + matched = False + out.append( + { + "turn_idx": idx, + "expected": expected, + "actual": actual, + "matched": matched, + "reason": reason, + } + ) + return out def score_fidelity(replay_results: list[dict]) -> float: @@ -345,13 +481,39 @@ def measure_token_cost( return rebuilt_tokens / pre_clear_tokens +def _run_subdir( + base: Path | None, + case: TranscriptCase, + trigger_threshold: float, + token_budget: int, +) -> Path | None: + """Compute a deterministic per-(case, config) subdirectory under `base`. + + Returns None if `base` is None — replay_post_fork falls back to its + placeholder mode in that case. The slug embeds threshold + budget so + a sweep produces one request file per (case, threshold, budget) cell + rather than overwriting a single shared file. + """ + if base is None: + return None + return base / f"{case.path.stem}__t{trigger_threshold}__b{token_budget}" + + def run_one( case: TranscriptCase, *, trigger_threshold: float, token_budget: int, + run_dir: Path | None = None, ) -> RunResult: - """End-to-end one run on one case at one config.""" + """End-to-end one run on one case at one config. + + When `run_dir` is provided, `replay_post_fork` writes a per-run + `replay_requests.jsonl` under `/__t__b/` + and joins any matching `replay_responses.jsonl`. Non-None default + is opt-in; bare callers (the v1.2.0 stub path) get unchanged + behaviour. + """ store = replay_to_fork(case) t0 = time.monotonic() rebuilt, latency_ms = run_rebuilder( @@ -360,7 +522,8 @@ def run_one( trigger_threshold=trigger_threshold, token_budget=token_budget, ) - replay = replay_post_fork(rebuilt, case) + case_run_dir = _run_subdir(run_dir, case, trigger_threshold, token_budget) + replay = replay_post_fork(rebuilt, case, run_dir=case_run_dir) fidelity = score_fidelity(replay) cost_ratio = measure_token_cost(rebuilt, case) return RunResult( @@ -384,9 +547,15 @@ def sweep_thresholds( *, thresholds: tuple[float, ...] = (0.5, 0.6, 0.7, 0.8, 0.9), token_budget: int = 2000, + run_dir: Path | None = None, ) -> SweepResult: runs = [ - run_one(case, trigger_threshold=t, token_budget=token_budget) + run_one( + case, + trigger_threshold=t, + token_budget=token_budget, + run_dir=run_dir, + ) for case in cases for t in thresholds ] @@ -399,9 +568,15 @@ def sweep_budgets( *, budgets: tuple[int, ...] = (500, 1000, 2000, 4000), trigger_threshold: float = 0.7, + run_dir: Path | None = None, ) -> SweepResult: runs = [ - run_one(case, trigger_threshold=trigger_threshold, token_budget=b) + run_one( + case, + trigger_threshold=trigger_threshold, + token_budget=b, + run_dir=run_dir, + ) for case in cases for b in budgets ] @@ -465,6 +640,19 @@ def main() -> int: p.add_argument("--mode", choices=("threshold-sweep", "budget-sweep", "dynamic", "regression"), required=True) p.add_argument("--corpus", type=Path, required=True) p.add_argument("--out", type=Path, required=True) + p.add_argument( + "--run-dir", + type=Path, + default=None, + help=( + "Optional base directory under which replay_post_fork writes " + "per-(case, config) replay_requests.jsonl and reads " + "replay_responses.jsonl. Without this flag the harness emits " + "placeholder rows (reason=needs_replay_client) and skips file " + "IO. See benchmarks/context-rebuilder/README.md for the " + "host-agent eval-replay flow (#600)." + ), + ) args = p.parse_args() cases = load_corpus(args.corpus) @@ -473,9 +661,9 @@ def main() -> int: return 1 if args.mode == "threshold-sweep": - result = sweep_thresholds(cases) + result = sweep_thresholds(cases, run_dir=args.run_dir) elif args.mode == "budget-sweep": - result = sweep_budgets(cases) + result = sweep_budgets(cases, run_dir=args.run_dir) elif args.mode == "dynamic": raise NotImplementedError("dynamic mode lands once the heuristic exists") elif args.mode == "regression": diff --git a/tests/test_context_rebuilder_eval_harness_wiring.py b/tests/test_context_rebuilder_eval_harness_wiring.py index b814bcda6..b4324fe24 100644 --- a/tests/test_context_rebuilder_eval_harness_wiring.py +++ b/tests/test_context_rebuilder_eval_harness_wiring.py @@ -302,6 +302,153 @@ def test_replay_post_fork_score_fidelity_returns_zero( assert harness.score_fidelity(results) == 0.0 +# --------------------------------------------------------------------- # +# replay_post_fork host-agent eval-replay (#600) # +# --------------------------------------------------------------------- # + + +def test_replay_post_fork_writes_requests_jsonl( + harness: ModuleType, transcript_path: Path, tmp_path: Path +) -> None: + """With `run_dir`, `replay_post_fork` writes one request row per + eval_turn to `/replay_requests.jsonl`.""" + case = harness.TranscriptCase( + path=transcript_path, task_type="debug", + fork_turn=2, eval_turns=(2, 4), + ) + run_dir = tmp_path / "run" + harness.replay_post_fork("REBUILT_BLOCK", case, run_dir=run_dir) + req_path = run_dir / harness.REPLAY_REQUESTS_FILENAME + assert req_path.is_file() + rows = [json.loads(line) for line in req_path.read_text().splitlines() if line.strip()] + assert len(rows) == 2 + by_idx = {r["turn_idx"]: r for r in rows} + assert set(by_idx) == {2, 4} + # Each row carries the rebuilt block plus the user prompt at-or-before + # the eval turn. + assert by_idx[2]["rebuilt_block"] == "REBUILT_BLOCK" + assert by_idx[2]["user_turn"] == "What about session-scoped retrieval?" + assert by_idx[2]["expected"] == "What about session-scoped retrieval?" + # eval_turn=4 is itself a user turn → user_turn == expected. + assert by_idx[4]["user_turn"] == "Does the floor apply to L0?" + + +def test_replay_post_fork_pending_reason_when_no_responses( + harness: ModuleType, transcript_path: Path, tmp_path: Path +) -> None: + """run_dir set but no response file → rows hold `pending_replay`, + distinct from the bare-stub `needs_replay_client`.""" + case = harness.TranscriptCase( + path=transcript_path, task_type="debug", + fork_turn=2, eval_turns=(2, 4), + ) + run_dir = tmp_path / "run" + rows = harness.replay_post_fork("rebuilt", case, run_dir=run_dir) + assert all(not r["matched"] for r in rows) + assert all(r["reason"] == harness.PENDING_REPLAY_REASON for r in rows) + + +def test_replay_pending_round_trip( + harness: ModuleType, transcript_path: Path, tmp_path: Path +) -> None: + """End-to-end: run replay_post_fork once to emit requests, hand-author + a `replay_responses.jsonl` (no child task), re-invoke replay_post_fork, + and assert score_fidelity reports the substring-match verdict. + + Acceptance bullet from #600. No model client involved — the round + trip is the contract; the dispatcher is operator-driven. + """ + case = harness.TranscriptCase( + path=transcript_path, task_type="debug", + fork_turn=2, eval_turns=(2, 4), + ) + run_dir = tmp_path / "run" + + # First pass: requests get written, responses absent → all pending. + first_pass = harness.replay_post_fork("rebuilt", case, run_dir=run_dir) + assert (run_dir / harness.REPLAY_REQUESTS_FILENAME).is_file() + assert all(r["reason"] == harness.PENDING_REPLAY_REASON for r in first_pass) + assert harness.score_fidelity(first_pass) == 0.0 + + # Hand-author replay_responses.jsonl: turn 2 substring-matches the + # expected text, turn 4 does not (drops to needs_llm_judge). + # Expected for idx=2 is the literal text at line 2: + # "What about session-scoped retrieval?" + responses_path = run_dir / harness.REPLAY_RESPONSES_FILENAME + responses_path.write_text( + json.dumps({ + "turn_idx": 2, + "actual": "Earlier we discussed: WHAT ABOUT SESSION-SCOPED RETRIEVAL? Yes.", + }) + "\n" + + json.dumps({ + "turn_idx": 4, + "actual": "no idea what the floor does, sorry", + }) + "\n", + encoding="utf-8", + ) + + # Second pass: same args, but now responses join in. + second_pass = harness.replay_post_fork("rebuilt", case, run_dir=run_dir) + by_idx = {r["turn_idx"]: r for r in second_pass} + assert by_idx[2]["matched"] is True + assert by_idx[2]["reason"] == "" + assert by_idx[2]["actual"].startswith("Earlier we discussed") + assert by_idx[4]["matched"] is False + assert by_idx[4]["reason"] == harness.NEEDS_LLM_JUDGE_REASON + assert by_idx[4]["actual"].startswith("no idea") + + # score_fidelity reports the substring-match-half verdict. + assert harness.score_fidelity(second_pass) == 0.5 + + +def test_replay_pending_partial_response_keeps_others_pending( + harness: ModuleType, transcript_path: Path, tmp_path: Path +) -> None: + """Acceptance: `missing rows stay pending_replay`. A response file + that only covers some eval_turns must not promote the others.""" + case = harness.TranscriptCase( + path=transcript_path, task_type="debug", + fork_turn=2, eval_turns=(2, 4), + ) + run_dir = tmp_path / "run" + harness.replay_post_fork("rebuilt", case, run_dir=run_dir) + (run_dir / harness.REPLAY_RESPONSES_FILENAME).write_text( + json.dumps({ + "turn_idx": 2, + # Expected at idx=2 is "What about session-scoped retrieval?"; + # actual contains that as a substring → matched=True. + "actual": "Reply: what about session-scoped retrieval? — covered.", + }) + "\n", + encoding="utf-8", + ) + rows = harness.replay_post_fork("rebuilt", case, run_dir=run_dir) + by_idx = {r["turn_idx"]: r for r in rows} + assert by_idx[2]["matched"] is True + assert by_idx[4]["matched"] is False + assert by_idx[4]["reason"] == harness.PENDING_REPLAY_REASON + assert by_idx[4]["actual"] == "" + + +def test_run_one_threads_run_dir_to_replay_post_fork( + harness: ModuleType, tmp_path: Path +) -> None: + """`run_one(run_dir=...)` produces a per-(case, config) subdirectory + under the base run_dir with a `replay_requests.jsonl` inside it.""" + corpus_dir = ( + _REPO_ROOT / "benchmarks" / "context-rebuilder" / "fixtures" / "synthetic" + ) + cases = harness.load_corpus(corpus_dir) + case = cases[0] + base = tmp_path / "sweep_run" + harness.run_one( + case, trigger_threshold=0.5, token_budget=1000, run_dir=base, + ) + expected_subdir = ( + base / f"{case.path.stem}__t0.5__b1000" + ) + assert (expected_subdir / harness.REPLAY_REQUESTS_FILENAME).is_file() + + # --------------------------------------------------------------------- # # End-to-end: threshold-sweep against the bundled synthetic fixture # # --------------------------------------------------------------------- #