From ddc1e2fd8a154ad0a6f5ad9524155d9b7f09a55b Mon Sep 17 00:00:00 2001 From: Philip John Basile Date: Wed, 26 Aug 2026 09:24:07 -0400 Subject: [PATCH] feat(replay): add capture plan orchestration --- .github/workflows/ci.yml | 2 +- mtplx/replay_orchestrator.py | 566 +++++++++++++++++++++++++++ scripts/bench_replay_orchestrator.py | 140 +++++++ tests/test_replay_orchestrator.py | 224 +++++++++++ 4 files changed, 931 insertions(+), 1 deletion(-) create mode 100644 mtplx/replay_orchestrator.py create mode 100644 scripts/bench_replay_orchestrator.py create mode 100644 tests/test_replay_orchestrator.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9ad5e86f6..6db9938b1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,5 +18,5 @@ jobs: python-version: "3.11" - run: python -m pip install -U pip - run: python -m pip install -e ".[dev,server]" - - run: python -m pytest tests/test_no_mlx_imports.py tests/test_public_cli.py tests/test_runtime_kpis.py tests/test_server_openai.py tests/test_openai_bridge.py tests/test_thermal.py tests/test_cache_state.py + - run: python -m pytest tests/test_no_mlx_imports.py tests/test_public_cli.py tests/test_runtime_kpis.py tests/test_server_openai.py tests/test_openai_bridge.py tests/test_thermal.py tests/test_cache_state.py tests/test_replay_orchestrator.py - run: python -m compileall -q mtplx tests diff --git a/mtplx/replay_orchestrator.py b/mtplx/replay_orchestrator.py new file mode 100644 index 000000000..cd83313ca --- /dev/null +++ b/mtplx/replay_orchestrator.py @@ -0,0 +1,566 @@ +"""Deterministic capture plan and replay receipt orchestration. + +This module owns bounded capture discovery, stable plan selection, stale-plan +validation, and atomic receipt storage. It deliberately does not execute a +candidate or evaluate outputs. Callers can connect any replay engine through +the mapping-only receipt boundary without importing that implementation here. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +import threading +import time +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass +from datetime import datetime +from pathlib import Path +from typing import Any + + +class ReplayPlanError(ValueError): + """Raised when a replay plan or receipt is invalid.""" + + +class StaleReplayPlanError(ReplayPlanError): + """Raised when capture evidence changes after plan creation.""" + + +@dataclass(frozen=True) +class CaptureFilter: + model: str | None = None + session_id: str | None = None + status: str | None = None + require_error: bool | None = None + minimum_prompt_tokens: int | None = None + maximum_prompt_tokens: int | None = None + created_after_s: float | None = None + created_before_s: float | None = None + + +@dataclass(frozen=True) +class ReplayPlanConfig: + maximum_scan_files: int = 5000 + maximum_cases: int = 128 + maximum_file_bytes: int = 4 * 1024 * 1024 + deterministic_seed: str = "mtplx" + deduplicate_public_fingerprints: bool = True + require_replayable_content: bool = False + + def __post_init__(self) -> None: + if self.maximum_scan_files < 1: + raise ValueError("maximum_scan_files must be at least 1") + if self.maximum_cases < 1: + raise ValueError("maximum_cases must be at least 1") + if self.maximum_file_bytes < 256: + raise ValueError("maximum_file_bytes must be at least 256") + + +@dataclass(frozen=True) +class CaptureEnvelope: + capture_id: str + path: str + file_sha256: str + public_fingerprint: str + request: Any | None + baseline_output: Any | None + metadata: Mapping[str, Any] + replayable: bool + unavailable_reason: str | None + + def to_dict(self, *, include_request: bool = False) -> dict[str, Any]: + payload = { + "capture_id": self.capture_id, + "path": self.path, + "file_sha256": self.file_sha256, + "public_fingerprint": self.public_fingerprint, + "metadata": dict(self.metadata), + "replayable": self.replayable, + "unavailable_reason": self.unavailable_reason, + } + if include_request: + payload["request"] = self.request + payload["baseline_output"] = self.baseline_output + return payload + + +@dataclass(frozen=True) +class ReplayPlan: + plan_id: str + created_at_s: float + capture_root: str + cases: tuple[CaptureEnvelope, ...] + skipped_count: int + duplicate_count: int + source_digest: str + config: ReplayPlanConfig + filters: CaptureFilter + + @property + def replayable_cases(self) -> int: + return sum(item.replayable for item in self.cases) + + def to_dict(self, *, include_requests: bool = False) -> dict[str, Any]: + return { + "schema_version": 1, + "plan_id": self.plan_id, + "created_at_s": self.created_at_s, + "capture_root": self.capture_root, + "cases": [ + item.to_dict(include_request=include_requests) for item in self.cases + ], + "case_count": len(self.cases), + "replayable_cases": self.replayable_cases, + "skipped_count": self.skipped_count, + "duplicate_count": self.duplicate_count, + "source_digest": self.source_digest, + "config": asdict(self.config), + "filters": asdict(self.filters), + "promotion_is_automatic": False, + } + + +@dataclass(frozen=True) +class ReplayReceipt: + receipt_id: str + plan_id: str + source_digest: str + candidate_name: str + created_at_s: float + report: Mapping[str, Any] + decision: Mapping[str, Any] + stale_check_passed: bool + promotion_applied: bool = False + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": 1, + "receipt_id": self.receipt_id, + "plan_id": self.plan_id, + "source_digest": self.source_digest, + "candidate_name": self.candidate_name, + "created_at_s": self.created_at_s, + "report": dict(self.report), + "decision": dict(self.decision), + "stale_check_passed": self.stale_check_passed, + "promotion_applied": self.promotion_applied, + } + + +def _canonical(value: Any) -> bytes: + return json.dumps( + value, + sort_keys=True, + ensure_ascii=False, + separators=(",", ":"), + default=str, + allow_nan=False, + ).encode("utf-8") + + +def _digest(value: Any) -> str: + return hashlib.sha256(_canonical(value)).hexdigest() + + +def _first(mapping: Mapping[str, Any], *paths: Sequence[str]) -> Any: + for path in paths: + value: Any = mapping + found = True + for part in path: + if not isinstance(value, Mapping) or part not in value: + found = False + break + value = value[part] + if found: + return value + return None + + +def _safe_int(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _safe_float(value: Any) -> float | None: + try: + return float(value) + except (TypeError, ValueError): + pass + if isinstance(value, str): + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + except ValueError: + pass + return None + + +def _public_fingerprint(record: Mapping[str, Any], file_sha256: str) -> str: + value = _first( + record, + ("request_fingerprint",), + ("request", "fingerprint"), + ("request", "sha256"), + ("prompt_sha256",), + ("prompt", "sha256"), + ) + if isinstance(value, str) and value: + return value + return file_sha256 + + +def _extract_replay_request(record: Mapping[str, Any]) -> tuple[Any | None, str | None]: + candidates = ( + _first(record, ("request", "payload")), + _first(record, ("request", "body")), + _first(record, ("request_payload",)), + _first(record, ("messages",)), + ) + for value in candidates: + if value is None: + continue + if isinstance(value, Mapping) and value.get("content_redacted") is True: + continue + if isinstance(value, Mapping) and set(value) <= { + "count", + "bytes", + "sha256", + "content_bytes", + "content_sha256", + "content_redacted", + }: + continue + return value, None + return None, "request_content_not_captured" + + +def _extract_baseline(record: Mapping[str, Any]) -> Any | None: + for value in ( + _first(record, ("outcome", "response")), + _first(record, ("outcome", "body")), + _first(record, ("response",)), + _first(record, ("response_payload",)), + ): + if value is None: + continue + if isinstance(value, Mapping) and value.get("content_redacted") is True: + continue + return value + return None + + +class ReplayOrchestrator: + """Build stable plans and persist engine-neutral replay receipts.""" + + def __init__( + self, + capture_root: str | os.PathLike[str], + *, + config: ReplayPlanConfig | None = None, + receipt_directory: str | os.PathLike[str] | None = None, + ) -> None: + self.capture_root = Path(capture_root).expanduser().resolve() + self.config = config or ReplayPlanConfig() + self.receipt_directory = ( + Path(receipt_directory).expanduser().resolve() + if receipt_directory is not None + else self.capture_root / "replay_receipts" + ) + self._last_plan: ReplayPlan | None = None + self._last_receipt: ReplayReceipt | None = None + self._lock = threading.RLock() + + def _capture_files(self) -> tuple[Path, ...]: + if not self.capture_root.exists(): + return () + rows: list[Path] = [] + for path in self.capture_root.rglob("*.json"): + if path.is_symlink() or not path.is_file(): + continue + try: + resolved = path.resolve(strict=True) + resolved.relative_to(self.capture_root) + except (OSError, ValueError): + continue + if self.receipt_directory in resolved.parents or "pruned" in resolved.parts: + continue + rows.append(resolved) + rows.sort(key=lambda path: path.as_posix()) + return tuple(rows[: self.config.maximum_scan_files]) + + def _read(self, path: Path) -> CaptureEnvelope | None: + try: + stat = path.stat() + if stat.st_size > self.config.maximum_file_bytes: + return None + raw = path.read_bytes() + record = json.loads(raw) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(record, Mapping): + return None + file_sha = hashlib.sha256(raw).hexdigest() + capture_id = str( + _first(record, ("request_id",), ("capture_id",), ("id",)) or path.stem + ) + request, unavailable = _extract_replay_request(record) + metadata = { + "model": _first(record, ("model",), ("request", "model")), + "session_id": _first(record, ("session_id",), ("request", "session_id")), + "status": _first(record, ("outcome", "status"), ("status",)), + "error_type": _first( + record, + ("outcome", "error_type"), + ("error", "type"), + ("error_type",), + ), + "prompt_tokens": _safe_int( + _first( + record, + ("prompt_tokens",), + ("request", "prompt_tokens"), + ("usage", "prompt_tokens"), + ) + ), + "created_at_s": _safe_float( + _first( + record, + ("created_at_s",), + ("timestamp_s",), + ("created_at",), + ) + ), + } + return CaptureEnvelope( + capture_id=capture_id, + path=str(path.relative_to(self.capture_root)), + file_sha256=file_sha, + public_fingerprint=_public_fingerprint(record, file_sha), + request=request, + baseline_output=_extract_baseline(record), + metadata=metadata, + replayable=request is not None, + unavailable_reason=unavailable, + ) + + @staticmethod + def _matches(envelope: CaptureEnvelope, filters: CaptureFilter) -> bool: + metadata = envelope.metadata + if filters.model is not None and metadata.get("model") != filters.model: + return False + if filters.session_id is not None and metadata.get("session_id") != filters.session_id: + return False + if filters.status is not None and str(metadata.get("status")) != filters.status: + return False + has_error = bool(metadata.get("error_type")) + if filters.require_error is not None and has_error != filters.require_error: + return False + prompt_tokens = _safe_int(metadata.get("prompt_tokens")) + if filters.minimum_prompt_tokens is not None and ( + prompt_tokens is None or prompt_tokens < filters.minimum_prompt_tokens + ): + return False + if filters.maximum_prompt_tokens is not None and ( + prompt_tokens is None or prompt_tokens > filters.maximum_prompt_tokens + ): + return False + created = _safe_float(metadata.get("created_at_s")) + if filters.created_after_s is not None and ( + created is None or created < filters.created_after_s + ): + return False + if filters.created_before_s is not None and ( + created is None or created > filters.created_before_s + ): + return False + return True + + def _rank_key(self, envelope: CaptureEnvelope) -> str: + return _digest( + { + "seed": self.config.deterministic_seed, + "capture": envelope.capture_id, + "fingerprint": envelope.public_fingerprint, + } + ) + + @staticmethod + def _source_digest(cases: Sequence[CaptureEnvelope]) -> str: + return _digest( + [ + { + "path": item.path, + "sha256": item.file_sha256, + "fingerprint": item.public_fingerprint, + } + for item in cases + ] + ) + + def _select( + self, filters: CaptureFilter + ) -> tuple[tuple[CaptureEnvelope, ...], int, int]: + skipped = 0 + duplicate_count = 0 + candidates: list[CaptureEnvelope] = [] + seen: set[str] = set() + for path in self._capture_files(): + envelope = self._read(path) + if envelope is None or not self._matches(envelope, filters): + skipped += 1 + continue + if self.config.require_replayable_content and not envelope.replayable: + skipped += 1 + continue + if ( + self.config.deduplicate_public_fingerprints + and envelope.public_fingerprint in seen + ): + duplicate_count += 1 + continue + seen.add(envelope.public_fingerprint) + candidates.append(envelope) + candidates.sort(key=lambda item: (self._rank_key(item), item.path)) + cases = tuple(candidates[: self.config.maximum_cases]) + return cases, skipped, duplicate_count + + def build_plan(self, filters: CaptureFilter | None = None) -> ReplayPlan: + selected_filter = filters or CaptureFilter() + cases, skipped, duplicate_count = self._select(selected_filter) + source_digest = self._source_digest(cases) + created = time.time() + plan_id = _digest( + { + "created_at_s": created, + "source_digest": source_digest, + "config": asdict(self.config), + "filters": asdict(selected_filter), + } + )[:24] + plan = ReplayPlan( + plan_id=plan_id, + created_at_s=created, + capture_root=str(self.capture_root), + cases=cases, + skipped_count=skipped, + duplicate_count=duplicate_count, + source_digest=source_digest, + config=self.config, + filters=selected_filter, + ) + with self._lock: + self._last_plan = plan + return plan + + def assert_fresh(self, plan: ReplayPlan) -> None: + if Path(plan.capture_root) != self.capture_root: + raise ReplayPlanError("plan capture root does not match orchestrator") + if plan.config != self.config: + raise ReplayPlanError("plan configuration does not match orchestrator") + current, _, _ = self._select(plan.filters) + current_digest = self._source_digest(current) + if current_digest != plan.source_digest: + raise StaleReplayPlanError("capture set changed after replay plan creation") + + @staticmethod + def _atomic_json(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + handle, temporary = tempfile.mkstemp( + prefix=path.name + ".", + suffix=".tmp", + dir=str(path.parent), + ) + try: + with os.fdopen(handle, "w", encoding="utf-8") as stream: + json.dump(payload, stream, sort_keys=True, ensure_ascii=False, indent=2) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + + def record_receipt( + self, + plan: ReplayPlan, + *, + candidate_name: str, + report: Mapping[str, Any], + decision: Mapping[str, Any], + ) -> ReplayReceipt: + """Validate source freshness and atomically persist an engine result.""" + if not candidate_name.strip(): + raise ReplayPlanError("candidate_name must not be empty") + self.assert_fresh(plan) + created = time.time() + receipt_payload = { + "plan_id": plan.plan_id, + "source_digest": plan.source_digest, + "candidate_name": candidate_name, + "created_at_s": created, + "report": dict(report), + "decision": dict(decision), + "stale_check_passed": True, + "promotion_applied": False, + } + receipt = ReplayReceipt( + receipt_id=_digest(receipt_payload)[:24], + **receipt_payload, + ) + self._atomic_json( + self.receipt_directory / f"{receipt.receipt_id}.json", + receipt.to_dict(), + ) + with self._lock: + self._last_receipt = receipt + return receipt + + def list_receipts(self, *, limit: int = 20) -> tuple[dict[str, Any], ...]: + if not self.receipt_directory.exists(): + return () + rows: list[dict[str, Any]] = [] + paths = sorted( + self.receipt_directory.glob("*.json"), + key=lambda path: path.stat().st_mtime, + reverse=True, + ) + for path in paths[: max(0, int(limit))]: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + if isinstance(payload, dict): + rows.append(payload) + return tuple(rows) + + def snapshot(self) -> dict[str, Any]: + with self._lock: + receipt_count = 0 + if self.receipt_directory.exists(): + try: + receipt_count = sum( + 1 for _ in self.receipt_directory.glob("*.json") + ) + except OSError: + receipt_count = 0 + return { + "available": True, + "enabled": self.capture_root.exists(), + "capture_root_configured": bool(str(self.capture_root)), + "maximum_cases": self.config.maximum_cases, + "content_capture_required": False, + "promotion_is_automatic": False, + "receipt_count": receipt_count, + "last_plan": self._last_plan.to_dict(include_requests=False) + if self._last_plan + else None, + "last_receipt": self._last_receipt.to_dict() + if self._last_receipt + else None, + } diff --git a/scripts/bench_replay_orchestrator.py b/scripts/bench_replay_orchestrator.py new file mode 100644 index 000000000..477fa456b --- /dev/null +++ b/scripts/bench_replay_orchestrator.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +"""Measure capture plan selection, freshness checks, and receipt persistence.""" + +from __future__ import annotations + +import argparse +import json +import statistics +import tempfile +import time +from pathlib import Path +from typing import Any + +from mtplx.replay_orchestrator import ( + ReplayOrchestrator, + ReplayPlanConfig, + StaleReplayPlanError, +) + + +def percentile(values: list[float], fraction: float) -> float: + ordered = sorted(values) + index = min(len(ordered) - 1, int(len(ordered) * fraction)) + return ordered[index] + + +def summary(samples_ns: list[int]) -> dict[str, float]: + samples_ms = [value / 1_000_000 for value in samples_ns] + return { + "mean_ms": round(statistics.fmean(samples_ms), 3), + "p50_ms": round(percentile(samples_ms, 0.50), 3), + "p95_ms": round(percentile(samples_ms, 0.95), 3), + "p99_ms": round(percentile(samples_ms, 0.99), 3), + } + + +def write_captures(root: Path, count: int) -> None: + for index in range(count): + payload = { + "request_id": f"case-{index:05d}", + "model": "benchmark-model", + "request_fingerprint": f"fingerprint-{index:05d}", + "prompt_tokens": 64 + index % 2048, + "request": {"payload": {"messages": index % 32}}, + } + (root / f"capture-{index:05d}.json").write_text( + json.dumps(payload, sort_keys=True, separators=(",", ":")), + encoding="utf-8", + ) + + +def run(args: argparse.Namespace) -> dict[str, Any]: + with tempfile.TemporaryDirectory(prefix="mtplx-replay-bench-") as temporary: + root = Path(temporary) + capture_root = root / "captures" + receipt_root = root / "receipts" + capture_root.mkdir() + write_captures(capture_root, args.captures) + orchestrator = ReplayOrchestrator( + capture_root, + config=ReplayPlanConfig( + maximum_scan_files=args.captures, + maximum_cases=args.maximum_cases, + deterministic_seed="benchmark", + ), + receipt_directory=receipt_root, + ) + + plan_samples: list[int] = [] + source_digests: set[str] = set() + plan = orchestrator.build_plan() + for _ in range(args.plan_iterations): + started = time.perf_counter_ns() + plan = orchestrator.build_plan() + plan_samples.append(time.perf_counter_ns() - started) + source_digests.add(plan.source_digest) + + freshness_samples: list[int] = [] + for _ in range(args.freshness_iterations): + started = time.perf_counter_ns() + orchestrator.assert_fresh(plan) + freshness_samples.append(time.perf_counter_ns() - started) + + receipt_samples: list[int] = [] + for index in range(args.receipt_iterations): + started = time.perf_counter_ns() + orchestrator.record_receipt( + plan, + candidate_name=f"candidate-{index}", + report={"pass_rate": 1.0, "case_count": len(plan.cases)}, + decision={"promote": True, "reasons": []}, + ) + receipt_samples.append(time.perf_counter_ns() - started) + + receipt_files = tuple(receipt_root.glob("*.json")) + temporary_files = tuple(receipt_root.glob("*.tmp")) + selected_capture = capture_root / plan.cases[0].path + selected_capture.write_text("{}", encoding="utf-8") + stale_started = time.perf_counter_ns() + stale_mutation_rejected = False + try: + orchestrator.assert_fresh(plan) + except StaleReplayPlanError: + stale_mutation_rejected = True + stale_rejection_ms = (time.perf_counter_ns() - stale_started) / 1_000_000 + return { + "captures": args.captures, + "selected_cases": len(plan.cases), + "plan_iterations": args.plan_iterations, + "freshness_iterations": args.freshness_iterations, + "receipt_iterations": args.receipt_iterations, + "plan_selection": summary(plan_samples), + "stale_plan_validation": summary(freshness_samples), + "atomic_receipt_write": summary(receipt_samples), + "stable_source_digest_count": len(source_digests), + "stale_mutation_rejected": stale_mutation_rejected, + "stale_rejection_ms": round(stale_rejection_ms, 3), + "receipt_files": len(receipt_files), + "temporary_files_remaining": len(temporary_files), + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--captures", type=int, default=1000) + parser.add_argument("--maximum-cases", type=int, default=128) + parser.add_argument("--plan-iterations", type=int, default=30) + parser.add_argument("--freshness-iterations", type=int, default=100) + parser.add_argument("--receipt-iterations", type=int, default=50) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + result = run(args) + encoded = json.dumps(result, indent=2, sort_keys=True) + print(encoded) + if args.output is not None: + args.output.write_text(encoded + "\n", encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/tests/test_replay_orchestrator.py b/tests/test_replay_orchestrator.py new file mode 100644 index 000000000..e301102ee --- /dev/null +++ b/tests/test_replay_orchestrator.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from mtplx.replay_orchestrator import ( + CaptureFilter, + ReplayOrchestrator, + ReplayPlanConfig, + ReplayPlanError, + StaleReplayPlanError, +) + + +def write_capture(root: Path, name: str, payload: dict) -> Path: + path = root / f"{name}.json" + path.write_text(json.dumps(payload), encoding="utf-8") + return path + + +def test_plan_selection_is_deterministic_deduplicated_and_bounded(tmp_path): + write_capture( + tmp_path, + "a", + { + "request_id": "a", + "model": "m", + "request": {"payload": {"prompt": "one"}}, + "request_fingerprint": "same", + "prompt_tokens": 4, + }, + ) + write_capture( + tmp_path, + "b", + { + "request_id": "b", + "model": "m", + "request": {"payload": {"prompt": "duplicate"}}, + "request_fingerprint": "same", + "prompt_tokens": 5, + }, + ) + write_capture( + tmp_path, + "c", + { + "request_id": "c", + "model": "m", + "request": {"payload": {"prompt": "two"}}, + "request_fingerprint": "different", + "prompt_tokens": 6, + }, + ) + orchestrator = ReplayOrchestrator( + tmp_path, + config=ReplayPlanConfig(maximum_cases=2, deterministic_seed="fixed"), + ) + first = orchestrator.build_plan(CaptureFilter(model="m")) + second = orchestrator.build_plan(CaptureFilter(model="m")) + assert [item.capture_id for item in first.cases] == [ + item.capture_id for item in second.cases + ] + assert len(first.cases) == 2 + assert first.duplicate_count == 1 + assert first.source_digest == second.source_digest + + +def test_redacted_count_only_capture_is_not_treated_as_replayable(tmp_path): + write_capture( + tmp_path, + "redacted", + { + "request_id": "redacted", + "prompt_tokens": 42, + "prompt": {"content_redacted": True, "content_sha256": "x" * 64}, + }, + ) + plan = ReplayOrchestrator(tmp_path).build_plan() + assert len(plan.cases) == 1 + assert plan.cases[0].replayable is False + assert plan.cases[0].unavailable_reason == "request_content_not_captured" + + +def test_stale_capture_is_rejected_before_receipt_write(tmp_path): + capture = write_capture( + tmp_path, + "case", + {"request_id": "case", "request": {"payload": {"prompt": "one"}}}, + ) + receipts = tmp_path / "receipts" + orchestrator = ReplayOrchestrator(tmp_path, receipt_directory=receipts) + plan = orchestrator.build_plan() + capture.write_text( + json.dumps( + {"request_id": "case", "request": {"payload": {"prompt": "changed"}}} + ), + encoding="utf-8", + ) + with pytest.raises(StaleReplayPlanError): + orchestrator.record_receipt( + plan, + candidate_name="candidate", + report={"passed": True}, + decision={"promote": True}, + ) + assert not receipts.exists() + + +def test_new_capture_that_changes_selection_makes_plan_stale(tmp_path): + write_capture(tmp_path, "first", {"request_id": "first"}) + orchestrator = ReplayOrchestrator(tmp_path) + plan = orchestrator.build_plan() + write_capture(tmp_path, "second", {"request_id": "second"}) + with pytest.raises(StaleReplayPlanError): + orchestrator.assert_fresh(plan) + + +def test_receipt_is_atomic_private_by_default_and_listable(tmp_path): + write_capture( + tmp_path, + "case", + { + "request_id": "case", + "request": {"payload": {"prompt": "private"}}, + "outcome": {"response": {"answer": "baseline"}}, + }, + ) + receipts = tmp_path / "receipts" + orchestrator = ReplayOrchestrator(tmp_path, receipt_directory=receipts) + plan = orchestrator.build_plan() + receipt = orchestrator.record_receipt( + plan, + candidate_name="candidate", + report={"pass_rate": 1.0}, + decision={"promote": True, "reasons": []}, + ) + files = list(receipts.glob("*.json")) + assert len(files) == 1 + assert list(receipts.glob("*.tmp")) == [] + raw = files[0].read_text(encoding="utf-8") + assert "private" not in raw + assert receipt.receipt_id in raw + assert receipt.promotion_applied is False + assert orchestrator.list_receipts(limit=1)[0]["receipt_id"] == receipt.receipt_id + + +def test_empty_candidate_name_is_rejected(tmp_path): + write_capture(tmp_path, "case", {"request_id": "case"}) + orchestrator = ReplayOrchestrator(tmp_path) + plan = orchestrator.build_plan() + with pytest.raises(ReplayPlanError, match="candidate_name"): + orchestrator.record_receipt( + plan, + candidate_name=" ", + report={}, + decision={}, + ) + + +def test_plan_from_another_capture_root_is_rejected(tmp_path): + source = tmp_path / "source" + other = tmp_path / "other" + source.mkdir() + other.mkdir() + write_capture(source, "case", {"request_id": "case"}) + plan = ReplayOrchestrator(source).build_plan() + with pytest.raises(ReplayPlanError, match="capture root"): + ReplayOrchestrator(other).assert_fresh(plan) + + +def test_filters_cover_error_and_prompt_size(tmp_path): + write_capture( + tmp_path, + "small", + { + "request_id": "small", + "prompt_tokens": 2, + "request": {"payload": {"prompt": "a"}}, + }, + ) + write_capture( + tmp_path, + "large-error", + { + "request_id": "large-error", + "prompt_tokens": 100, + "error_type": "RuntimeError", + "request": {"payload": {"prompt": "b"}}, + }, + ) + plan = ReplayOrchestrator(tmp_path).build_plan( + CaptureFilter(require_error=True, minimum_prompt_tokens=50) + ) + assert [item.capture_id for item in plan.cases] == ["large-error"] + + +def test_iso_timestamp_filter_and_external_symlink_are_handled_safely(tmp_path): + write_capture( + tmp_path, + "iso", + { + "request_id": "iso", + "created_at": "2026-08-24T00:00:00Z", + "request": {"payload": {"prompt": "safe"}}, + }, + ) + outside = tmp_path.parent / "outside-capture.json" + outside.write_text( + json.dumps( + {"request_id": "outside", "request": {"payload": {"prompt": "outside"}}} + ), + encoding="utf-8", + ) + link = tmp_path / "outside.json" + try: + link.symlink_to(outside) + except OSError: + pass + orchestrator = ReplayOrchestrator(tmp_path) + plan = orchestrator.build_plan(CaptureFilter(created_after_s=1_700_000_000)) + assert [item.capture_id for item in plan.cases] == ["iso"]