diff --git a/config/repo_review_automation.toml b/config/repo_review_automation.toml index 1de484da2..8a8cc88fe 100644 --- a/config/repo_review_automation.toml +++ b/config/repo_review_automation.toml @@ -78,12 +78,16 @@ repairs derived GitNexus corruption by quarantining .gitnexus and requiring a clean forced rebuild at the exact reviewed commit. A Claude session-capacity message with a concrete reset time is preserved beside the agent log and deferred until that reset without consuming the normal retry budget; parent -phase timeouts include that bounded wait. The coordinator returns failure if +phase timeouts include that bounded wait. If origin/main advances after round +1, later-phase retries cannot repair stale findings: the coordinator preserves +and quarantines the repo's entire analysis, then restarts at round 1 so the +GitNexus map and every finding share the new exact-head provenance. Those full +repo restarts use the same bounded repair budget. The coordinator returns failure if round-1, round-2, body-writer, the final evaluator, or required docs-drift publication evidence remains failed. """ output = "docs/reports/repo-review/round2/__/converged.json" -if_failed = "The phase is repaired and retried twice. A persistent failure remains visible in state.json and makes the coordinator exit nonzero." +if_failed = "The phase is repaired and retried twice. Head drift also refreshes the active-repo preflight and restarts from round 1, using the same budget (up to 3x per-repo wall time). A persistent failure remains visible in state.json and makes the coordinator exit nonzero." [operational_notes.queue_builder] step = "3. Queue-builder preview (does NOT write the artifact)" diff --git a/docs/ops/REPO_REVIEW_PROCESS.md b/docs/ops/REPO_REVIEW_PROCESS.md index 7564a1b1c..5ab700718 100644 --- a/docs/ops/REPO_REVIEW_PROCESS.md +++ b/docs/ops/REPO_REVIEW_PROCESS.md @@ -79,6 +79,17 @@ round-1 work, but quarantines existing round-2 turn outputs because a malformed partial JSON file would otherwise be reused on every retry. A body-writer retry receives the exact deterministic validator errors from the prior failed attempt and must pass the body/path validator, not only the JSON schema. If a +later phase discovers that `origin/main` advanced after round 1, the +coordinator does not retry stale findings. It moves all repo-scoped round-1 and +round-2 outputs into the repair directory, copies the repo's coordinator logs +there as evidence, and restarts the repository at round 1. The stale per-repo +preflight brief is also quarantined and the active-repo preflight is regenerated +before round 1. The fresh run re-syncs the checkout, refreshes or +rebuilds GitNexus, and establishes a new exact-head provenance chain. Full +repo restarts are bounded by the same `--repair-attempts` budget. This can +increase the worst-case per-repo wall time by up to three times at the default +budget, but keeps the recovery limit aligned with phase repair and avoids a +second operator-tuned control. If a required round-1, round-2, or body-writer phase exhausts its repair budget, the coordinator aborts immediately before later repos or aggregate producers run. It writes `repo-review-run-failure.json` with the failed repo, phase, timestamp, diff --git a/scripts/repo_review_coordinator.py b/scripts/repo_review_coordinator.py index a252d2432..e8ae138c4 100644 --- a/scripts/repo_review_coordinator.py +++ b/scripts/repo_review_coordinator.py @@ -39,6 +39,7 @@ import subprocess import sys import tomllib +from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass from datetime import UTC, datetime @@ -600,6 +601,7 @@ def run_subprocess_with_repairs( repo: str, output_dir: Path, repair_attempts: int, + stop_retry_when: Callable[[Path], bool] | None = None, ) -> tuple[StepResult, list[dict[str, Any]], list[dict[str, Any]]]: """Run a repo phase and perform a recorded, bounded repair before retry.""" attempts: list[dict[str, Any]] = [] @@ -623,6 +625,8 @@ def run_subprocess_with_repairs( ) if result.succeeded or attempt_number > repair_attempts: break + if stop_retry_when is not None and stop_retry_when(log_path): + break try: repair = prepare_phase_retry( phase=name, @@ -656,6 +660,101 @@ def run_subprocess_with_repairs( return result, attempts, repairs +HEAD_DRIFT_MARKERS = ( + "does not match origin/main", + "exact-head mismatch", + "exact head mismatch", + "source commit mismatch", + "source_commit mismatch", +) + + +def log_indicates_head_drift(log_path: Path) -> bool: + """Return whether a failed phase used a source head that is now stale.""" + try: + diagnostic = log_path.read_text(encoding="utf-8", errors="replace").lower() + except OSError: + return False + return any(marker in diagnostic for marker in HEAD_DRIFT_MARKERS) + + +def prepare_head_drift_restart( + *, + repo: str, + output_dir: Path, + log_dir: Path, + restart_number: int, + failed_phase: str, +) -> dict[str, Any]: + """Quarantine all repo-scoped analysis before restarting from round 1. + + A later phase cannot repair findings produced from an obsolete commit. A + full repo restart deliberately discards only derived analysis for that repo; + the next round-1 sync then establishes a new exact-head provenance chain and + rebuilds GitNexus when necessary. + """ + safe = repo.replace("/", "__") + timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%S.%fZ") + repair_dir = output_dir / "repairs" / safe / f"{timestamp}-head-drift-{restart_number}" + repair_dir.mkdir(parents=True, exist_ok=False) + preserved: list[str] = [] + quarantined: list[str] = [] + + repo_log_dir = log_dir / safe + if repo_log_dir.is_dir(): + log_destination = repair_dir / "coordinator-logs" + shutil.copytree(repo_log_dir, log_destination) + preserved.append(str(log_destination)) + + round1_root = output_dir / "round1" + for agent_dir in sorted(round1_root.iterdir()) if round1_root.is_dir() else []: + source = agent_dir / safe + if not source.exists(): + continue + destination = repair_dir / "round1" / agent_dir.name + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(source), str(destination)) + quarantined.append(str(destination)) + + repo_round2 = output_dir / "round2" / safe + if repo_round2.exists(): + destination = repair_dir / "round2" + shutil.move(str(repo_round2), str(destination)) + quarantined.append(str(destination)) + + repo_preflight = output_dir / "repos" / safe + if repo_preflight.exists(): + destination = repair_dir / "preflight" + shutil.move(str(repo_preflight), str(destination)) + quarantined.append(str(destination)) + + actions = { + "phase": "head-drift-restart", + "failed_phase": failed_phase, + "repo": repo, + "restart_number": restart_number, + "created_at": datetime.now(UTC).isoformat(), + "repair_dir": str(repair_dir), + "preserved": preserved, + "quarantined": quarantined, + } + (repair_dir / "repair.json").write_text( + json.dumps(actions, indent=2) + "\n", + encoding="utf-8", + ) + + state = load_state(output_dir, repo) + attempt = begin_attempt(state, phase="head-drift-restart", agent="coordinator") + finish_attempt( + state, + attempt, + succeeded=True, + notes=f"restart {restart_number}: stale {failed_phase} analysis moved to {repair_dir}", + ) + save_state(output_dir, state) + return actions + + def failed_repo_phase(report: dict[str, Any]) -> str | None: """Return the first failed required phase, or ``None`` for a valid repo result.""" if not (report.get("round1") or {}).get("succeeded"): @@ -895,6 +994,10 @@ def coordinate_repo( repo=repo, output_dir=output_dir, repair_attempts=repair_attempts, + # Restoring the same converged baseline cannot cure source-head drift. + # Return control immediately so the outer loop can restart this repo at + # round 1 and re-establish exact-head provenance. + stop_retry_when=log_indicates_head_drift, ) report["body_writer"] = { "succeeded": bw_result.succeeded, @@ -907,6 +1010,131 @@ def coordinate_repo( return report +def coordinate_repo_with_restarts( + *, + repo: str, + output_dir: Path, + workflows_steward_root: Path, + registry_path: Path, + agents: list[str], + log_dir: Path, + round1_timeout: int, + round2_timeout: int, + max_turns: int, + skip_gate_enabled: bool, + repair_attempts: int = 2, +) -> dict[str, Any]: + """Coordinate a repo, restarting the full provenance chain on head drift.""" + restarts: list[dict[str, Any]] = [] + prior_attempts: list[dict[str, Any]] = [] + report: dict[str, Any] = {} + + for restart_number in range(repair_attempts + 1): + report = coordinate_repo( + repo=repo, + output_dir=output_dir, + workflows_steward_root=workflows_steward_root, + registry_path=registry_path, + agents=agents, + log_dir=log_dir, + round1_timeout=round1_timeout, + round2_timeout=round2_timeout, + max_turns=max_turns, + skip_gate_enabled=skip_gate_enabled, + repair_attempts=repair_attempts, + ) + failed_phase = failed_repo_phase(report) + if failed_phase is None: + break + + phase_log_name = { + "round-1": "round1-runner.log", + "round-2": "round2-runner.log", + "body-writer": "body-writer.log", + }[failed_phase] + phase_log = log_dir / repo.replace("/", "__") / phase_log_name + if not log_indicates_head_drift(phase_log) or restart_number >= repair_attempts: + break + + prior_attempts.append(report) + try: + repair = prepare_head_drift_restart( + repo=repo, + output_dir=output_dir, + log_dir=log_dir, + restart_number=restart_number + 1, + failed_phase=failed_phase, + ) + except OSError as exc: + report["head_drift_restart_error"] = f"repair preparation failed: {exc}" + print( + f"[coordinator] {repo}: head-drift restart preparation failed: {exc}", + file=sys.stderr, + ) + break + restarts.append(repair) + + # The review brief is source-derived too. Regenerate active-repo inputs + # after quarantine so round 1 cannot consume an inventory or GitNexus + # status captured from the obsolete commit. Aggregate files written by + # this preflight are provisional and will be replaced by the final pass. + restart_preflight_log = ( + log_dir / repo.replace("/", "__") / f"preflight-restart-{restart_number + 1}.log" + ) + restart_preflight = run_subprocess( + [ + sys.executable, + str(workflows_steward_root / "scripts" / "repo_review_evaluator.py"), + "--registry", + str(registry_path), + "--output-dir", + str(output_dir), + "--status", + "active", + "--skip-gitnexus-preflight", + ], + cwd=workflows_steward_root, + log_path=restart_preflight_log, + name="head-drift-preflight", + timeout=1200, + ) + repair["preflight"] = { + "succeeded": restart_preflight.succeeded, + "duration_seconds": restart_preflight.duration_seconds, + "notes": restart_preflight.notes, + } + try: + (Path(repair["repair_dir"]) / "repair.json").write_text( + json.dumps(repair, indent=2) + "\n", + encoding="utf-8", + ) + except OSError as exc: + report["head_drift_restart_error"] = f"repair evidence update failed: {exc}" + print( + f"[coordinator] {repo}: head-drift repair evidence update failed: {exc}", + file=sys.stderr, + ) + break + if not restart_preflight.succeeded: + report["head_drift_restart_error"] = restart_preflight.notes + print( + f"[coordinator] {repo}: head-drift preflight refresh failed: " + f"{restart_preflight.notes}", + file=sys.stderr, + ) + break + print( + f"[coordinator] {repo}: {failed_phase} detected source-head drift; " + f"restart {restart_number + 1} recorded at {repair['repair_dir']}; " + "restarting from round 1" + ) + + if restarts: + report["head_drift_restarts"] = restarts + report["prior_stale_head_attempts"] = prior_attempts + return report + + # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- @@ -999,7 +1227,7 @@ def run(args: argparse.Namespace) -> int: reports: list[dict[str, Any]] = [] for repo in target_repos: print(f"[coordinator] {repo}: starting per-repo flow") - report = coordinate_repo( + report = coordinate_repo_with_restarts( repo=repo, output_dir=output_dir, workflows_steward_root=workflows_steward_root, diff --git a/tests/scripts/test_repo_review_coordinator.py b/tests/scripts/test_repo_review_coordinator.py index a0a2cd2e9..13e8dee54 100644 --- a/tests/scripts/test_repo_review_coordinator.py +++ b/tests/scripts/test_repo_review_coordinator.py @@ -770,6 +770,337 @@ def fail_repair(**_kwargs): assert repairs[0]["succeeded"] is False +def test_body_writer_head_drift_returns_control_for_full_repo_restart( + tmp_path: Path, monkeypatch +) -> None: + calls = 0 + + def fail_with_head_drift(_cmd, *, cwd, log_path, name, timeout): + nonlocal calls + calls += 1 + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "sync check failed: HEAD (abc) does not match origin/main\n", + encoding="utf-8", + ) + return coordinator.StepResult( + name=name, succeeded=False, duration_seconds=0.01, notes="exit 1" + ) + + monkeypatch.setattr(coordinator, "run_subprocess", fail_with_head_drift) + result, attempts, repairs = coordinator.run_subprocess_with_repairs( + ["body-writer"], + cwd=tmp_path, + log_path=tmp_path / "body-writer.log", + name="body-writer", + timeout=30, + repo="stranske/Example", + output_dir=tmp_path / "out", + repair_attempts=2, + stop_retry_when=coordinator.log_indicates_head_drift, + ) + + assert result.succeeded is False + assert calls == 1 + assert len(attempts) == 1 + assert repairs == [] + + +def test_coordinate_repo_restarts_round1_after_late_head_drift(tmp_path: Path, monkeypatch) -> None: + repo = "stranske/Example" + safe = "stranske__Example" + output_dir = tmp_path / "review" + log_dir = output_dir / "logs" / "coordinator" + calls = 0 + preflight_calls = 0 + + def fake_run_subprocess(cmd, *, cwd, log_path, name, timeout): + nonlocal preflight_calls + assert name == "head-drift-preflight" + assert cmd[cmd.index("--status") + 1] == "active" + preflight_calls += 1 + review_inputs = output_dir / "repos" / safe / "review-inputs.md" + review_inputs.parent.mkdir(parents=True, exist_ok=True) + review_inputs.write_text("fresh head\n", encoding="utf-8") + return coordinator.StepResult(name=name, succeeded=True, duration_seconds=0.01) + + def fake_coordinate_repo(**_kwargs): + nonlocal calls + calls += 1 + phase_log = log_dir / safe / "body-writer.log" + phase_log.parent.mkdir(parents=True, exist_ok=True) + round1 = output_dir / "round1" / "codex" / safe / "findings.json" + round1.parent.mkdir(parents=True, exist_ok=True) + round1.write_text(f'{{"attempt": {calls}}}\n', encoding="utf-8") + round2 = output_dir / "round2" / safe / "converged.json" + round2.parent.mkdir(parents=True, exist_ok=True) + round2.write_text(f'{{"attempt": {calls}}}\n', encoding="utf-8") + if calls == 1: + review_inputs = output_dir / "repos" / safe / "review-inputs.md" + review_inputs.parent.mkdir(parents=True, exist_ok=True) + review_inputs.write_text("stale head\n", encoding="utf-8") + phase_log.write_text( + "HEAD (old) does not match origin/main\n", + encoding="utf-8", + ) + return { + "repo": repo, + "round1": {"succeeded": True}, + "round2": {"succeeded": True}, + "body_writer": {"succeeded": False}, + "skip_gate_fired": False, + } + assert (output_dir / "repos" / safe / "review-inputs.md").read_text() == "fresh head\n" + phase_log.write_text("complete\n", encoding="utf-8") + return { + "repo": repo, + "round1": {"succeeded": True}, + "round2": {"succeeded": True}, + "body_writer": {"succeeded": True}, + "skip_gate_fired": False, + } + + monkeypatch.setattr(coordinator, "coordinate_repo", fake_coordinate_repo) + monkeypatch.setattr(coordinator, "run_subprocess", fake_run_subprocess) + report = coordinator.coordinate_repo_with_restarts( + repo=repo, + output_dir=output_dir, + workflows_steward_root=tmp_path, + registry_path=tmp_path / "config" / "repo_review_registry.json", + agents=["codex", "claude"], + log_dir=log_dir, + round1_timeout=30, + round2_timeout=30, + max_turns=1, + skip_gate_enabled=False, + repair_attempts=2, + ) + + assert calls == 2 + assert preflight_calls == 1 + assert report["body_writer"]["succeeded"] is True + assert len(report["head_drift_restarts"]) == 1 + repair_dir = Path(report["head_drift_restarts"][0]["repair_dir"]) + assert (repair_dir / "round1" / "codex" / "findings.json").is_file() + assert (repair_dir / "round2" / "converged.json").is_file() + assert (repair_dir / "preflight" / "review-inputs.md").read_text() == "stale head\n" + assert ( + json.loads((output_dir / "round1" / "codex" / safe / "findings.json").read_text())[ + "attempt" + ] + == 2 + ) + + +def test_head_drift_restart_io_failure_returns_controlled_report( + tmp_path: Path, monkeypatch +) -> None: + repo = "stranske/Example" + log_dir = tmp_path / "review" / "logs" / "coordinator" + phase_log = log_dir / "stranske__Example" / "body-writer.log" + phase_log.parent.mkdir(parents=True) + phase_log.write_text("HEAD (old) does not match origin/main\n", encoding="utf-8") + failed_report = { + "repo": repo, + "round1": {"succeeded": True}, + "round2": {"succeeded": True}, + "body_writer": {"succeeded": False}, + "skip_gate_fired": False, + } + monkeypatch.setattr(coordinator, "coordinate_repo", lambda **_kwargs: failed_report.copy()) + monkeypatch.setattr( + coordinator, + "prepare_head_drift_restart", + lambda **_kwargs: (_ for _ in ()).throw(OSError("repair volume unavailable")), + ) + + report = coordinator.coordinate_repo_with_restarts( + repo=repo, + output_dir=tmp_path / "review", + workflows_steward_root=tmp_path, + registry_path=tmp_path / "config" / "repo_review_registry.json", + agents=["codex", "claude"], + log_dir=log_dir, + round1_timeout=30, + round2_timeout=30, + max_turns=1, + skip_gate_enabled=False, + repair_attempts=2, + ) + + assert report["body_writer"]["succeeded"] is False + assert "repair volume unavailable" in report["head_drift_restart_error"] + + +def test_persistent_head_drift_exhausts_bounded_repo_restarts(tmp_path: Path, monkeypatch) -> None: + repo = "stranske/Example" + safe = "stranske__Example" + output_dir = tmp_path / "review" + log_dir = output_dir / "logs" / "coordinator" + calls = 0 + + def always_drift(**_kwargs): + nonlocal calls + calls += 1 + phase_log = log_dir / safe / "body-writer.log" + phase_log.parent.mkdir(parents=True, exist_ok=True) + phase_log.write_text("exact-head mismatch\n", encoding="utf-8") + converged = output_dir / "round2" / safe / "converged.json" + converged.parent.mkdir(parents=True, exist_ok=True) + converged.write_text(f'{{"attempt": {calls}}}\n', encoding="utf-8") + return { + "repo": repo, + "round1": {"succeeded": True}, + "round2": {"succeeded": True}, + "body_writer": {"succeeded": False}, + "skip_gate_fired": False, + } + + monkeypatch.setattr(coordinator, "coordinate_repo", always_drift) + monkeypatch.setattr( + coordinator, + "run_subprocess", + lambda _cmd, **kwargs: coordinator.StepResult( + name=kwargs["name"], succeeded=True, duration_seconds=0.01 + ), + ) + report = coordinator.coordinate_repo_with_restarts( + repo=repo, + output_dir=output_dir, + workflows_steward_root=tmp_path, + registry_path=tmp_path / "config" / "repo_review_registry.json", + agents=["codex", "claude"], + log_dir=log_dir, + round1_timeout=30, + round2_timeout=30, + max_turns=1, + skip_gate_enabled=False, + repair_attempts=2, + ) + + assert calls == 3 + assert report["body_writer"]["succeeded"] is False + assert len(report["head_drift_restarts"]) == 2 + assert len(report["prior_stale_head_attempts"]) == 2 + + +def test_head_drift_preflight_failure_stops_before_stale_round1_restart( + tmp_path: Path, monkeypatch +) -> None: + repo = "stranske/Example" + log_dir = tmp_path / "review" / "logs" / "coordinator" + phase_log = log_dir / "stranske__Example" / "body-writer.log" + phase_log.parent.mkdir(parents=True) + calls = 0 + + def stale_attempt(**_kwargs): + nonlocal calls + calls += 1 + phase_log.write_text("source commit mismatch\n", encoding="utf-8") + return { + "repo": repo, + "round1": {"succeeded": True}, + "round2": {"succeeded": True}, + "body_writer": {"succeeded": False}, + "skip_gate_fired": False, + } + + monkeypatch.setattr(coordinator, "coordinate_repo", stale_attempt) + monkeypatch.setattr( + coordinator, + "run_subprocess", + lambda _cmd, **kwargs: coordinator.StepResult( + name=kwargs["name"], + succeeded=False, + duration_seconds=0.01, + notes="exit 1; preflight unavailable", + ), + ) + report = coordinator.coordinate_repo_with_restarts( + repo=repo, + output_dir=tmp_path / "review", + workflows_steward_root=tmp_path, + registry_path=tmp_path / "config" / "repo_review_registry.json", + agents=["codex", "claude"], + log_dir=log_dir, + round1_timeout=30, + round2_timeout=30, + max_turns=1, + skip_gate_enabled=False, + repair_attempts=2, + ) + + assert calls == 1 + assert report["body_writer"]["succeeded"] is False + assert "preflight unavailable" in report["head_drift_restart_error"] + + +def test_run_writes_failure_marker_after_head_drift_restart_exhaustion( + tmp_path: Path, monkeypatch +) -> None: + registry_path = tmp_path / "config" / "repo_review_registry.json" + registry_path.parent.mkdir(parents=True) + registry_path.write_text("{}\n", encoding="utf-8") + output_dir = tmp_path / "out" + output_dir.mkdir() + for name in coordinator.AGGREGATE_OUTPUT_NAMES: + (output_dir / name).write_text("stale\n", encoding="utf-8") + monkeypatch.setattr( + coordinator, + "load_registry", + lambda _path: ( + tmp_path, + [], + [SimpleNamespace(repo="stranske/Example", status="active")], + [], + ), + ) + calls: list[str] = [] + + def fake_run_subprocess(_cmd, *, cwd, log_path, name, timeout): + calls.append(name) + log_path.parent.mkdir(parents=True, exist_ok=True) + if name == "body-writer": + log_path.write_text("HEAD (old) does not match origin/main\n", encoding="utf-8") + return coordinator.StepResult( + name=name, succeeded=False, duration_seconds=0.01, notes="exit 1" + ) + log_path.write_text(f"{name} complete\n", encoding="utf-8") + return coordinator.StepResult(name=name, succeeded=True, duration_seconds=0.01) + + monkeypatch.setattr(coordinator, "run_subprocess", fake_run_subprocess) + args = SimpleNamespace( + output_dir=str(output_dir), + registry=str(registry_path), + repos=[], + agents=["codex", "claude"], + skip_preflight=True, + skip_gitnexus_preflight=True, + round1_timeout=30, + round2_timeout=30, + max_turns=1, + repair_attempts=1, + docs_drift_timeout=30, + disable_skip_gate=True, + skip_auto_archive=True, + ) + + assert coordinator.run(args) == 1 + assert calls == [ + "round-1", + "round-2", + "body-writer", + "head-drift-preflight", + "round-1", + "round-2", + "body-writer", + ] + failure = json.loads((output_dir / "repo-review-run-failure.json").read_text()) + assert failure["phase"] == "body-writer" + assert len(failure["report"]["head_drift_restarts"]) == 1 + assert all(not (output_dir / name).exists() for name in coordinator.AGGREGATE_OUTPUT_NAMES) + + def test_run_returns_nonzero_after_body_writer_repairs_exhausted( tmp_path: Path, monkeypatch ) -> None: