diff --git a/memory/specs/anu-system-spec-changelog.md b/memory/specs/anu-system-spec-changelog.md index 0c900dd3..b6d57271 100644 --- a/memory/specs/anu-system-spec-changelog.md +++ b/memory/specs/anu-system-spec-changelog.md @@ -1,3 +1,11 @@ +## [2026-06-04 06:15] 변경 내역 +- [cron] `cron` 데이터 변경 +- [task_stats] `task_stats` 데이터 변경 + +## [2026-06-04 00:16] 변경 내역 +- [cron] `cron` 데이터 변경 +- [task_stats] `task_stats` 데이터 변경 + ## [2026-05-07 00:15] 변경 내역 - [cron] `cron` 데이터 변경 - [services] `running_services` 추가: `infokeyword-worker.service` diff --git a/memory/specs/anu-system-spec.md b/memory/specs/anu-system-spec.md index 6545df1b..6c8383f4 100644 --- a/memory/specs/anu-system-spec.md +++ b/memory/specs/anu-system-spec.md @@ -1125,11 +1125,11 @@ finish-task.sh → .done 파일 생성 → task-timer 종료 → notify-completi | 항목 | 값 | |------|-----| -| 총 작업 수 | 1867 | -| 완료된 작업 | 1842 | -| 완료율 | 98.7% | -| 평균 소요시간 | 43분 37초 | -| 최대 소요시간 | 100시간 49분 39초 | +| 총 작업 수 | 1942 | +| 완료된 작업 | 1898 | +| 완료율 | 97.7% | +| 평균 소요시간 | 59분 20초 | +| 최대 소요시간 | 512시간 25분 20초 | | 최소 소요시간 | 0초 | diff --git a/tests/regression/test_replacement_pr_runner_2510.py b/tests/regression/test_replacement_pr_runner_2510.py index cb01bc6d..e69de29b 100644 --- a/tests/regression/test_replacement_pr_runner_2510.py +++ b/tests/regression/test_replacement_pr_runner_2510.py @@ -1,493 +0,0 @@ -"""task-2510 회귀 테스트 — replacement_pr_runner 12 케이스. -QA 담당: 모리건(Morrigan) -대상: utils/replacement_pr_runner.py -""" -from __future__ import annotations - -import json -import subprocess -import sys -from dataclasses import asdict -from pathlib import Path - -import pytest - -WORKSPACE = Path(__file__).resolve().parent.parent.parent -if str(WORKSPACE) in sys.path: - sys.path.remove(str(WORKSPACE)) -sys.path.insert(0, str(WORKSPACE)) - -from utils.replacement_pr_runner import ( # noqa: E402 # pyright: ignore[reportMissingImports] - ReplacementPRRunner, - assert_no_cherry_pick, - detect_contamination, - transplant_expected_files, - assert_clean_working_tree, - precheck_local_replacement_diff, -) -from utils.automation_contracts import ( # noqa: E402 # pyright: ignore[reportMissingImports] - ReplacementResult, - CriticalEscalationType, -) -from utils.merge_queue_executor import ( # noqa: E402 # pyright: ignore[reportMissingImports] - TaskSpec, - assert_no_forbidden_git_flags, -) - - -# ─── helpers ────────────────────────────────────────────────────────────────── - -def cp(returncode=0, stdout="", stderr=""): - return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr) - - -def make_runner(returns_by_args=None, *, default_returncode=0, default_stdout=""): - """fake runner — args 패턴 매칭으로 응답 inject.""" - calls = [] - routes = list(returns_by_args.items()) if returns_by_args else [] - - def runner(args, cwd=None, timeout=60): - calls.append({"args": list(args), "cwd": cwd, "timeout": timeout}) - joined = " ".join(str(a) for a in args) - for tokens, response in routes: - if all(tok in joined for tok in tokens): - return response - return cp(returncode=default_returncode, stdout=default_stdout) - - setattr(runner, "calls", calls) - return runner - - -def make_spec(task_id="task-2510", expected=None, dependency=None) -> TaskSpec: - return TaskSpec( - task_id=task_id, - expected_files=expected or [ - "utils/replacement_pr_runner.py", - "tests/regression/test_replacement_pr_runner_2510.py", - ], - risk_area="replacement_pr", - dependency=dependency or [], - parallel_policy="serial_only", - merge_queue_position=7, - stale_recheck_required=True, - cherry_pick_allowed=False, - ) - - -def fake_pr_meta(head_ref="task/task-2507-dev5", head_sha="abcdef1", task_id="task-2507", files=None, **_extra): - """fake pr metadata. **_extra: positional pr_number 등 호출자 ergonomics 흡수.""" - return { - "head_ref": head_ref, - "head_sha": head_sha, - "base_ref": "main", - "task_id": task_id, - "files": files or [], - "title": f"[{task_id}] test fixture", - "number": _extra.get("pr_number", 60), - } - - -# ─── T01 clean PR → no-op ───────────────────────────────────────────────── -def test_t01_clean_pr_no_op(monkeypatch): - spec = make_spec() - expected = list(spec.expected_files) - runner = make_runner({}) - rpr = ReplacementPRRunner(runner=runner, dry_run=False) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(files=expected)) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: list(expected)) - - result = rpr.execute(60, task_spec=spec) - assert isinstance(result, ReplacementResult) - assert result.success is True - assert result.replacement_pr is None - assert result.original_pr_preserved is True - assert sorted(result.effective_diff_files) == sorted(expected) - assert result.forbidden_paths == [] - assert result.failure_reason is None - - -# ─── T02 contaminated detection ─────────────────────────────────────────── -def test_t02_contaminated_detection(): - expected = ["utils/replacement_pr_runner.py", "tests/regression/test_replacement_pr_runner_2510.py"] - contaminated_extra = expected + ["utils/rogue_extra_module.py", "scripts/unrelated_script.sh"] - result = detect_contamination(contaminated_extra, expected) - assert result["contaminated"] is True - assert "utils/rogue_extra_module.py" in result["extra"] - assert "scripts/unrelated_script.sh" in result["extra"] - - -# ─── T03 forbidden path → Critical FORBIDDEN_PATH_INTRUSION ────────────── -def test_t03_forbidden_path_intrusion(monkeypatch): - spec = make_spec() - expected = list(spec.expected_files) - forbidden_file = ".github/workflows/ci.yml" - - runner = make_runner({}) - rpr = ReplacementPRRunner(runner=runner, dry_run=False) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(files=expected + [forbidden_file])) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: expected + [forbidden_file]) - - result = rpr.execute(60, task_spec=spec) - assert result.success is False - assert result.failure_reason is not None - assert CriticalEscalationType.FORBIDDEN_PATH_INTRUSION.value in result.failure_reason - assert forbidden_file in result.forbidden_paths - - -# ─── T04 transplant: expected_files만 git show로 이식 (★ tmp_path 격리) ── -def test_t04_transplant_expected_files_uses_git_show(tmp_path): - """★ tmp_path 격리: 실제 source 손상 방지.""" - expected = ["utils/sample.py", "tests/regression/test_sample.py"] - head_sha = "abcdef1234" - runner = make_runner( - {("git", "show"): cp(returncode=0, stdout="# fake transplanted content\n")}, - default_returncode=0, - ) - - transplant_expected_files( - expected, - head_sha, - runner, - repo_dir=str(tmp_path), # ★ tmp_path 격리 - ) - - # git show 호출 확인 - runner_calls = getattr(runner, "calls") - show_calls = [c for c in runner_calls if "git" in c["args"] and "show" in c["args"]] - assert len(show_calls) >= len(expected) - - # cherry-pick 호출 없음 - cherry_pick_calls = [c for c in runner_calls if "cherry-pick" in " ".join(str(a) for a in c["args"])] - assert not cherry_pick_calls - - # tmp_path 안에만 파일이 작성됨 (실제 source 손상 X) - for f in expected: - assert (tmp_path / f).exists() - assert "fake transplanted content" in (tmp_path / f).read_text() - - -# ─── T05 원 PR 보존 + [REPLACED] 코멘트 (close/delete 호출 X) ──────────── -def test_t05_original_pr_preserved_comment_posted(monkeypatch, tmp_path): - spec = make_spec() - expected = list(spec.expected_files) - contaminated = expected + ["utils/rogue_extra_module.py"] - - runner = make_runner({}, default_returncode=0, default_stdout="") - rpr = ReplacementPRRunner(runner=runner, dry_run=False, repo_dir=str(tmp_path)) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(pr_number=60, files=contaminated)) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: contaminated) - monkeypatch.setattr(mod, "assert_clean_working_tree", lambda r, repo_dir=None: None) - monkeypatch.setattr(mod, "create_clean_replacement_branch", lambda task_id, r, *, timestamp=None, repo_dir=None: "task/task-2510-replacement-20260508") - monkeypatch.setattr(mod, "transplant_expected_files", lambda exp, src, r, *, repo_dir=None: list(exp)) - monkeypatch.setattr(mod, "commit_local", lambda task_id, r, *, repo_dir=None: cp(returncode=0)) - monkeypatch.setattr(mod, "precheck_local_replacement_diff", lambda branch, exp, r, *, repo_dir=None: (True, [], [])) - monkeypatch.setattr(mod, "push_branch", lambda branch, r, *, repo_dir=None: cp(returncode=0)) - monkeypatch.setattr(mod, "open_replacement_pr", lambda task_id, branch, source_pr, r, *, repo_dir=None: 61) - monkeypatch.setattr(mod, "validate_replacement_diff", lambda replacement_pr, exp, r: (True, [], [])) - - # post_replaced_comment는 stub하지 않고 실제 호출 → runner.calls에 기록됨 - result = rpr.execute(60, task_spec=spec) - - runner_calls = getattr(runner, "calls") - all_calls_joined = [" ".join(str(a) for a in c["args"]) for c in runner_calls] - - # 금지 호출 없음 - for call_str in all_calls_joined: - assert "gh pr close" not in call_str - assert not ("gh pr edit" in call_str and "--state" in call_str and "closed" in call_str) - assert not ("gh api" in call_str and "DELETE" in call_str) - assert not ("git push" in call_str and "--delete" in call_str) - - # [REPLACED] 코멘트 호출 있음 - comment_calls = [c for c in all_calls_joined if "gh pr comment" in c and "[REPLACED]" in c] - assert comment_calls, f"[REPLACED] comment must be posted, got calls: {all_calls_joined}" - assert result.original_pr_preserved is True - assert result.success is True - assert result.replacement_pr == 61 - - -# ─── T06 validate_replacement_diff: tuple return ────────────────────────── -def test_t06_validate_replacement_diff_exact_match(): - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - expected = ["utils/replacement_pr_runner.py", "tests/regression/test_replacement_pr_runner_2510.py"] - - files_payload_ok = json.dumps({"files": [{"path": p} for p in expected]}) - runner_ok = make_runner({("gh", "pr", "view"): cp(returncode=0, stdout=files_payload_ok)}) - valid, extra, missing = mod.validate_replacement_diff(61, expected, runner_ok) - assert valid is True - assert extra == [] - assert missing == [] - - extra_files_list = expected + ["utils/rogue_extra.py"] - files_payload_extra = json.dumps({"files": [{"path": p} for p in extra_files_list]}) - runner_extra = make_runner({("gh", "pr", "view"): cp(returncode=0, stdout=files_payload_extra)}) - valid_e, extra_e, _missing_e = mod.validate_replacement_diff(61, expected, runner_extra) - assert valid_e is False - assert "utils/rogue_extra.py" in extra_e - - -# ─── T07 replacement 실패 → Critical ────────────────────────────────────── -def test_t07_replacement_failure_critical(monkeypatch, tmp_path): - spec = make_spec() - expected = list(spec.expected_files) - contaminated = expected + ["utils/rogue.py"] - - runner = make_runner({}, default_returncode=0) - rpr = ReplacementPRRunner(runner=runner, dry_run=False, repo_dir=str(tmp_path)) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(files=contaminated)) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: contaminated) - # 실제 실행 경로(push_branch)에서 RuntimeError 시뮬레이션 - def boom(*a, **k): - raise RuntimeError("PUSH_FAILED simulated") - monkeypatch.setattr(mod, "assert_clean_working_tree", lambda *a, **k: None) - monkeypatch.setattr(mod, "create_clean_replacement_branch", lambda *a, **k: "task/task-2510-replacement-zzz") - monkeypatch.setattr(mod, "transplant_expected_files", lambda *a, **k: list(expected)) - monkeypatch.setattr(mod, "commit_local", lambda *a, **k: cp(returncode=0)) - monkeypatch.setattr(mod, "precheck_local_replacement_diff", lambda *a, **k: (True, [], [])) - monkeypatch.setattr(mod, "push_branch", boom) # ★ 실제 사용 함수에 모킹 - - result = rpr.execute(60, task_spec=spec) - assert result.success is False - assert result.failure_reason in { - CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value, - CriticalEscalationType.REPLACEMENT_PR_FAILED.value, - } - - -# ─── T08 PR #54 fixture 78건 → contaminated ──────────────────────────────── -def test_t08_pr54_fixture_78_files_contaminated(): - expected = ["utils/replacement_pr_runner.py", "tests/regression/test_replacement_pr_runner_2510.py"] - contaminated_files = list(expected) - for prefix in ["task-2487+1", "task-2503", "task-2485+1", "task-2488", "task-2489", "task-2493"]: - for i in range(10): - contaminated_files.append(f"utils/{prefix}_module_{i}.py") - contaminated_files.append(f"tests/regression/test_{prefix}_{i}.py") - for i in range(3): - contaminated_files.append(f"scripts/poc_script_{i}.sh") - contaminated_files.append(f"tests/POC/poc_test_{i}.py") - contaminated_files = list(dict.fromkeys(contaminated_files)) - assert len(contaminated_files) >= 78 # 회장 명시 78건 이상 확보 - - result = detect_contamination(contaminated_files, expected) - assert result["contaminated"] is True - assert len(result["extra"]) >= 1 - - -# ─── T09 task-2506 (117건) / task-2507 fixture ─────────────────────────────── -def test_t09_task2506_117_files_contaminated(): - expected_2506 = ["utils/auto_gemini_triage.py", "tests/regression/test_auto_gemini_triage_2506.py"] - contaminated_2506 = list(expected_2506) - for i in range(57): - contaminated_2506.append(f"utils/task-2479-dev1_accumulated_{i}.py") - contaminated_2506.append(f"tests/regression/test_task2479_{i}.py") - contaminated_2506.append("scripts/dev1_bootstrap.sh") - contaminated_2506.append("tests/POC/poc_gemini.py") - contaminated_2506 = list(dict.fromkeys(contaminated_2506)) - r2506 = detect_contamination(contaminated_2506, expected_2506) - assert r2506["contaminated"] is True - - expected_2507 = ["utils/replacement_pr_runner.py", "tests/regression/test_replacement_pr_runner_2510.py"] - contaminated_2507 = list(expected_2507) - for i in range(39): - contaminated_2507.append(f"utils/task-2507_base_acc_{i}.py") - contaminated_2507.append(f"tests/regression/test_task2507_acc_{i}.py") - contaminated_2507 = list(dict.fromkeys(contaminated_2507)) - r2507 = detect_contamination(contaminated_2507, expected_2507) - assert r2507["contaminated"] is True - - -# ─── T10 자동 cherry-pick 금지 ──────────────────────────────────────────── -def test_t10_assert_no_cherry_pick_raises(): - with pytest.raises(RuntimeError) as excinfo: - assert_no_cherry_pick(["git", "cherry-pick", "abc"]) - assert "CHERRY_PICK_FORBIDDEN" in str(excinfo.value) - - -def test_t10_assert_no_cherry_pick_safe_merge(): - # merge는 차단되지 않음 - assert_no_cherry_pick(["git", "merge", "abc"]) # no exception - assert_no_cherry_pick(["git", "show", "abc:file.py"]) # no exception - - -# ─── T11 force/rebase/admin flag 정적 차단 ──────────────────────────────── -def test_t11_force_flag_raises(): - with pytest.raises(RuntimeError): - assert_no_forbidden_git_flags(["git", "push", "--force"]) - - -def test_t11_admin_flag_raises(): - with pytest.raises(RuntimeError): - assert_no_forbidden_git_flags(["gh", "pr", "merge", "--admin"]) - - -def test_t11_rebase_raises(): - with pytest.raises(RuntimeError): - assert_no_forbidden_git_flags(["git", "rebase", "main"]) - - -# ─── T12 ReplacementResult JSON 직렬화 ──────────────────────────────────── -def test_t12_replacement_result_json_roundtrip(): - # 성공 - ok = ReplacementResult( - source_pr=54, replacement_pr=61, original_pr_preserved=True, - expected_files=["utils/replacement_pr_runner.py", "tests/regression/test_replacement_pr_runner_2510.py"], - effective_diff_files=["utils/replacement_pr_runner.py", "tests/regression/test_replacement_pr_runner_2510.py"], - forbidden_paths=[], success=True, failure_reason=None, - ) - payload_ok = json.dumps(asdict(ok)) - parsed_ok = json.loads(payload_ok) - rebuilt_ok = ReplacementResult(**parsed_ok) - assert rebuilt_ok.success is True - assert rebuilt_ok.replacement_pr == 61 - - # 실패 - fail = ReplacementResult( - source_pr=54, replacement_pr=None, original_pr_preserved=True, - expected_files=["utils/replacement_pr_runner.py"], effective_diff_files=[], - forbidden_paths=[], success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value, - ) - payload_fail = json.dumps(asdict(fail)) - parsed_fail = json.loads(payload_fail) - rebuilt_fail = ReplacementResult(**parsed_fail) - assert rebuilt_fail.success is False - assert rebuilt_fail.failure_reason == CriticalEscalationType.REPLACEMENT_PR_FAILED.value - - -# ─── T13 pre-flight dirty tree → REPLACEMENT_PR_AUTO_CREATION_FAILED ────── -def test_t13_dirty_working_tree_fails(monkeypatch, tmp_path): - """dirty working tree 감지 시 replacement 흐름 진입 전에 실패 반환.""" - spec = make_spec() - expected = list(spec.expected_files) - contaminated = expected + ["utils/rogue_dirty.py"] - - # git status --porcelain 이 비어있지 않은 응답 반환 → dirty - dirty_runner = make_runner( - {("git", "status", "--porcelain"): cp(returncode=0, stdout=" M utils/dirty_file.py\n")}, - default_returncode=0, - default_stdout="", - ) - rpr = ReplacementPRRunner(runner=dirty_runner, dry_run=False, repo_dir=str(tmp_path)) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(pr_number=60, files=contaminated)) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: contaminated) - - result = rpr.execute(60, task_spec=spec) - - assert result.success is False - assert result.failure_reason == CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value - # original PR은 건드리지 않음 - assert result.original_pr_preserved is True - # escalation packet 채워짐 - assert rpr.last_escalation_packet is not None - assert rpr.last_escalation_packet.escalation_type == CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF - - -# ─── T14 precheck_local_replacement_diff mismatch → PR open 호출 X ────── -def test_t14_precheck_mismatch_no_pr_open(monkeypatch, tmp_path): - """로컬 diff 사전 검증 실패 시 gh pr create 호출 없이 REPLACEMENT_PR_FAILED 반환.""" - spec = make_spec() - expected = list(spec.expected_files) - contaminated = expected + ["utils/rogue_precheck.py"] - - # status --porcelain은 clean (empty) - # precheck_local_replacement_diff 에서 mismatch 반환을 monkeypatch - runner = make_runner({}, default_returncode=0, default_stdout="") - rpr = ReplacementPRRunner(runner=runner, dry_run=False, repo_dir=str(tmp_path)) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(pr_number=60, files=contaminated)) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: contaminated) - monkeypatch.setattr(mod, "assert_clean_working_tree", lambda r, repo_dir=None: None) - monkeypatch.setattr(mod, "create_clean_replacement_branch", lambda task_id, r, *, timestamp=None, repo_dir=None: "task/task-2510-replacement-precheck") - monkeypatch.setattr(mod, "transplant_expected_files", lambda exp, src, r, *, repo_dir=None: list(exp)) - monkeypatch.setattr(mod, "commit_local", lambda task_id, r, *, repo_dir=None: cp(returncode=0)) - # precheck 실패 (extra 파일이 있음) - monkeypatch.setattr(mod, "precheck_local_replacement_diff", lambda branch, exp, r, *, repo_dir=None: (False, ["utils/unexpected_extra.py"], [])) - - # open_replacement_pr가 호출되면 테스트 실패하도록 - pr_open_called = [] - def fail_if_called(*a, **k): - pr_open_called.append(True) - return 99 - monkeypatch.setattr(mod, "open_replacement_pr", fail_if_called) - - result = rpr.execute(60, task_spec=spec) - - assert result.success is False - assert result.failure_reason == CriticalEscalationType.REPLACEMENT_PR_FAILED.value - assert not pr_open_called, "gh pr create must NOT be called when precheck fails" - # escalation packet 채워짐 - assert rpr.last_escalation_packet is not None - assert rpr.last_escalation_packet.escalation_type == CriticalEscalationType.REPLACEMENT_PR_FAILED - - -# ─── T15 build_escalation_packet 흐름 — 실패 시 last_escalation_packet ── -def test_t15_escalation_packet_populated_on_failure(monkeypatch, tmp_path): - """실패 경로에서 runner.last_escalation_packet이 채워지는지 검증.""" - spec = make_spec() - expected = list(spec.expected_files) - contaminated = expected + ["utils/rogue_escalation.py"] - - runner = make_runner({}, default_returncode=0, default_stdout="") - rpr = ReplacementPRRunner(runner=runner, dry_run=False, repo_dir=str(tmp_path)) - - import utils.replacement_pr_runner as mod # pyright: ignore[reportMissingImports] - monkeypatch.setattr(mod, "fetch_pr_metadata", lambda pr, r: fake_pr_meta(pr_number=60, files=contaminated)) - monkeypatch.setattr(mod, "compute_effective_diff", lambda meta, r: contaminated) - monkeypatch.setattr(mod, "assert_clean_working_tree", lambda r, repo_dir=None: None) - monkeypatch.setattr(mod, "create_clean_replacement_branch", lambda task_id, r, *, timestamp=None, repo_dir=None: "task/task-2510-replacement-esc") - monkeypatch.setattr(mod, "transplant_expected_files", lambda exp, src, r, *, repo_dir=None: list(exp)) - # commit_local에서 RuntimeError 발생 → REPLACEMENT_PR_AUTO_CREATION_FAILED - def boom_commit(*a, **k): - raise RuntimeError("COMMIT_FAILED simulated for T15") - monkeypatch.setattr(mod, "commit_local", boom_commit) - - # 초기에는 None - assert rpr.last_escalation_packet is None - - result = rpr.execute(60, task_spec=spec) - - assert result.success is False - # last_escalation_packet이 채워졌는지 - assert rpr.last_escalation_packet is not None - pkt = rpr.last_escalation_packet - assert pkt.pr_number == 60 - assert pkt.task_id == spec.task_id - assert pkt.escalation_type in ( - CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, - CriticalEscalationType.REPLACEMENT_PR_FAILED, - ) - assert pkt.reason != "" - assert isinstance(pkt.evidence, dict) - - -# ─── T16 wiring 활성화 회귀 (task-2516) ────────────────────────────────────── -def test_wiring_activated_default_runtime_path_2516(): - """task-2516: replacement_pr_runner 모듈의 top-level import에서 merge_queue_executor와의 - circular import가 제거되어 default wiring path(_WIRING_AVAILABLE=True)가 활성화되는지 회귀 검증. - """ - import importlib - # Reset state to trigger fresh import (test isolation). - for mod_name in ("utils.merge_queue_executor", "utils.replacement_pr_runner"): - if mod_name in sys.modules: - del sys.modules[mod_name] - mqe = importlib.import_module("utils.merge_queue_executor") - assert mqe._WIRING_AVAILABLE is True, ( - "W1 wiring 비활성 — replacement_pr_runner.py에 circular import가 남아있음" - ) - assert mqe.ReplacementPRRunner is not None, "ReplacementPRRunner가 None — wiring 누락" - # task-2510에서 정의된 wrapper 함수들도 lazy import 경로로 정상 호출 가능해야 함 - rpr = importlib.import_module("utils.replacement_pr_runner") - equal, extra, missing = rpr.compare_effective_diff(["a.py"], ["a.py"]) - assert equal is True - assert extra == [] and missing == [] diff --git a/utils/replacement_pr_runner.py b/utils/replacement_pr_runner.py index 1fa8b2d2..e69de29b 100644 --- a/utils/replacement_pr_runner.py +++ b/utils/replacement_pr_runner.py @@ -1,718 +0,0 @@ -"""utils/replacement_pr_runner.py — task-2510 5 모듈 #2. -회장 명시: contaminated PR 자동 감지 + origin/main 기준 clean branch에서 -expected_files만 이식하여 replacement PR 자동 생성. -원 PR은 close/delete 없이 보존. -""" -from __future__ import annotations - -import argparse -import json -import logging -import os -import re -import subprocess -import sys -from dataclasses import asdict -from datetime import datetime, timezone -from pathlib import Path -from typing import Any, Callable, Optional, TYPE_CHECKING, TypeAlias - -WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) - -# CLI 직접 실행 시 패키지 루트를 sys.path에 추가 -_HERE = Path(__file__).resolve().parent.parent # utils/ → worktree root -if str(_HERE) not in sys.path: - sys.path.insert(0, str(_HERE)) - -from utils.automation_contracts import ( # noqa: E402 # pyright: ignore[reportMissingImports] - ReplacementResult, - CriticalEscalationType, - EscalationPacket, -) - -# ─── circular import 회피 (task-2516) ──────────────────────────────────────── -# merge_queue_executor.py가 top-level에서 ReplacementPRRunner를 import하기 때문에 -# 본 모듈이 top-level에서 merge_queue_executor를 import하면 cycle이 발생한다. -# wrapper 함수로 lazy 위임하여 default wiring path(_WIRING_AVAILABLE=True)를 활성화한다. - -TaskSpec: TypeAlias = Any # runtime placeholder — 실제 클래스는 type hint 용도로만 참조됨 -if TYPE_CHECKING: - from utils.merge_queue_executor import TaskSpec # type: ignore[no-redef] # noqa: F401 # pyright: ignore[reportMissingImports] - -def compare_effective_diff(*args, **kwargs): - from utils import merge_queue_executor as _mqe # pyright: ignore[reportMissingImports] - return _mqe.compare_effective_diff(*args, **kwargs) - - -def detect_forbidden_paths(*args, **kwargs): - from utils import merge_queue_executor as _mqe # pyright: ignore[reportMissingImports] - return _mqe.detect_forbidden_paths(*args, **kwargs) - - -def assert_no_forbidden_git_flags(*args, **kwargs): - from utils import merge_queue_executor as _mqe # pyright: ignore[reportMissingImports] - return _mqe.assert_no_forbidden_git_flags(*args, **kwargs) - - -def load_task_spec(*args, **kwargs): - from utils import merge_queue_executor as _mqe # pyright: ignore[reportMissingImports] - return _mqe.load_task_spec(*args, **kwargs) - -logger = logging.getLogger(__name__) -RunnerType = Callable[..., subprocess.CompletedProcess] - - -def _default_runner(args, cwd=None, timeout=60): - return subprocess.run(args, cwd=cwd or str(WORKSPACE), capture_output=True, text=True, timeout=timeout) - - -# ─── §1 cherry-pick 정적 차단 ──────────────────────────────────────────────── -def assert_no_cherry_pick(args: list[str]) -> None: - """args에 'cherry-pick' 토큰이 들어가면 RuntimeError(CHERRY_PICK_FORBIDDEN).""" - if "cherry-pick" in args: - raise RuntimeError("CHERRY_PICK_FORBIDDEN") - - -# ─── §2 PR metadata 수집 ──────────────────────────────────────────────────── -def fetch_pr_metadata(pr_number: int, runner: RunnerType) -> dict: - """gh pr view --json headRefName,headRefOid,baseRefName,files,title. - return {"head_ref": ..., "head_sha": ..., "base_ref": ..., "files": [...], "task_id": ..., "title": ..., "number": pr_number} - 실패 시 RuntimeError. - task_id는 title 또는 head_ref에서 [task-NNNN] 패턴 추출. - """ - args = ["gh", "pr", "view", str(pr_number), "--json", - "headRefName,headRefOid,baseRefName,files,title"] - result = runner(args) - if result.returncode != 0: - raise RuntimeError(f"FETCH_PR_METADATA_FAILED: pr={pr_number} stderr={result.stderr!r}") - data = json.loads(result.stdout or "{}") - head_ref = data.get("headRefName", "") - title = data.get("title", "") - files = [f.get("path", "") for f in (data.get("files") or []) if f.get("path")] - # task_id 추출 - m = re.search(r"task-\d+(?:\+\d+)?", title) or re.search(r"task-\d+(?:\+\d+)?", head_ref) - task_id = m.group(0) if m else "unknown" - return { - "head_ref": head_ref, - "head_sha": data.get("headRefOid", ""), - "base_ref": data.get("baseRefName", "main"), - "files": files, - "task_id": task_id, - "title": title, - "number": pr_number, - } - - -# ─── §3 effective diff 산출 ───────────────────────────────────────────────── -def compute_effective_diff(pr_meta: dict, runner: RunnerType) -> list[str]: - """gh pr view에서 받은 files 우선 사용, 비어있으면 git diff origin/main...PR_HEAD --name-only. - """ - if pr_meta.get("files"): - return list(pr_meta["files"]) - head_sha = pr_meta.get("head_sha", "") - if not head_sha: - return [] - args = ["git", "diff", "--name-only", f"origin/{pr_meta.get('base_ref', 'main')}...{head_sha}"] - assert_no_forbidden_git_flags(args) - result = runner(args) - if result.returncode != 0: - return [] - return [line.strip() for line in (result.stdout or "").splitlines() if line.strip()] - - -# ─── §4 contaminated 판정 ────────────────────────────────────────────────── -def detect_contamination( - effective_files: list[str], - expected_files: list[str], - extra_forbidden_patterns: Optional[list] = None, -) -> dict: - """반환 dict: {"contaminated": bool, "forbidden_paths": [...], "extra": [...], "missing": [...]}""" - extra_pats = None - if extra_forbidden_patterns: - extra_pats = [p if hasattr(p, "search") else re.compile(p) for p in extra_forbidden_patterns] - forbidden = detect_forbidden_paths(effective_files, expected_files, extra_patterns=extra_pats) - equal, extra, missing = compare_effective_diff(effective_files, expected_files) - contaminated = bool(forbidden) or not equal - return {"contaminated": contaminated, "forbidden_paths": forbidden, "extra": extra, "missing": missing} - - -# ─── §5 clean replacement branch 생성 ─────────────────────────────────────── -def create_clean_replacement_branch( - task_id: str, - runner: RunnerType, - *, - timestamp: Optional[str] = None, - repo_dir: Optional[str] = None, -) -> str: - """origin/main 기준 신규 brace `task/-replacement-` 생성. - fetch origin main으로 stale base 회피. force/rebase/cherry-pick 금지. - """ - if timestamp is None: - timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") - branch = f"task/{task_id}-replacement-{timestamp}" - - # ★ Codex high #1: stale origin/main 회피를 위해 fetch 먼저 수행 - fetch_args = ["git", "fetch", "origin", "main", "--quiet"] - assert_no_forbidden_git_flags(fetch_args) - assert_no_cherry_pick(fetch_args) - fetch_result = runner(fetch_args, cwd=repo_dir) - if fetch_result.returncode != 0: - raise RuntimeError( - f"FETCH_ORIGIN_MAIN_FAILED: stderr={fetch_result.stderr!r}" - ) - - args = ["git", "checkout", "-b", branch, "origin/main"] - assert_no_forbidden_git_flags(args) - assert_no_cherry_pick(args) - result = runner(args, cwd=repo_dir) - if result.returncode != 0: - raise RuntimeError(f"CREATE_CLEAN_BRANCH_FAILED: branch={branch} stderr={result.stderr!r}") - return branch - - -# ─── §6 expected_files 이식 (cherry-pick 금지) ───────────────────────────── -def transplant_expected_files( - expected_files: list[str], - source_head: str, - runner: RunnerType, - *, - repo_dir: Optional[str] = None, -) -> list[str]: - """각 파일 git show : → write → git add. cherry-pick 정적 차단. - 파일 시스템에 실제로 write하므로 호출자는 임시 작업 dir(tmp_path 등)을 repo_dir로 줘야 source 손상 방지. - """ - transplanted: list[str] = [] - cwd = repo_dir or str(WORKSPACE) - for filepath in expected_files: - show_args = ["git", "show", f"{source_head}:{filepath}"] - assert_no_forbidden_git_flags(show_args) - assert_no_cherry_pick(show_args) - sr = runner(show_args, cwd=cwd) - if sr.returncode != 0: - raise RuntimeError(f"GIT_SHOW_FAILED: file={filepath} sha={source_head} stderr={sr.stderr!r}") - target = Path(cwd) / filepath - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(sr.stdout, encoding="utf-8") - add_args = ["git", "add", filepath] - assert_no_forbidden_git_flags(add_args) - assert_no_cherry_pick(add_args) - ar = runner(add_args, cwd=cwd) - if ar.returncode != 0: - raise RuntimeError(f"GIT_ADD_FAILED: file={filepath} stderr={ar.stderr!r}") - transplanted.append(filepath) - return transplanted - - -# ─── §7 commit (push X) ──────────────────────────────────────────────────── -def commit_local( - task_id: str, - runner: RunnerType, - *, - repo_dir: Optional[str] = None, -): - """git commit -m only (push 하지 않음). force 금지. 실패 시 RuntimeError.""" - cwd = repo_dir or str(WORKSPACE) - commit_msg = f"[{task_id}] replacement: expected_files only (auto-generated)" - cargs = ["git", "commit", "-m", commit_msg] - assert_no_forbidden_git_flags(cargs) - assert_no_cherry_pick(cargs) - cr = runner(cargs, cwd=cwd) - if cr.returncode != 0: - raise RuntimeError(f"COMMIT_FAILED: stderr={cr.stderr!r}") - return cr # CompletedProcess of commit - - -# ─── §7b push branch ─────────────────────────────────────────────────────── -def push_branch( - branch: str, - runner: RunnerType, - *, - repo_dir: Optional[str] = None, -): - """git push origin . force 금지. 실패 시 RuntimeError.""" - cwd = repo_dir or str(WORKSPACE) - pargs = ["git", "push", "origin", branch] - assert_no_forbidden_git_flags(pargs) - assert_no_cherry_pick(pargs) - pr = runner(pargs, cwd=cwd) - if pr.returncode != 0: - raise RuntimeError(f"PUSH_FAILED: stderr={pr.stderr!r}") - return pr - - -# ─── §7c commit_and_push (하위 호환) ────────────────────────────────────── -def commit_and_push( - task_id: str, - branch: str, - runner: RunnerType, - *, - repo_dir: Optional[str] = None, - push: bool = True, -): - """git commit -m + (선택적) git push. force 금지. 실패 시 RuntimeError.""" - cr = commit_local(task_id, runner, repo_dir=repo_dir) - if push: - push_branch(branch, runner, repo_dir=repo_dir) - return cr # CompletedProcess of commit - - -# ─── §8 로컬 diff 사전 검증 (push 전) ──────────────────────────────────── -def precheck_local_replacement_diff( - branch: str, - expected_files: list[str], - runner: RunnerType, - *, - repo_dir: Optional[str] = None, -) -> tuple[bool, list[str], list[str]]: - """commit 후 push 전, 로컬 git diff 기반 사전 검증. - - git show --stat HEAD 또는 git diff --name-only origin/main...HEAD 호출. - return (valid, extra, missing). - valid=True이면 expected_files와 일치하고 forbidden path 없음. - """ - cwd = repo_dir or str(WORKSPACE) - # git diff --name-only origin/main...HEAD - diff_args = ["git", "diff", "--name-only", "origin/main...HEAD"] - assert_no_forbidden_git_flags(diff_args) - assert_no_cherry_pick(diff_args) - result = runner(diff_args, cwd=cwd) - if result.returncode != 0: - # fallback: git show --stat HEAD - show_args = ["git", "show", "--stat", "--name-only", "--format=", "HEAD"] - assert_no_forbidden_git_flags(show_args) - assert_no_cherry_pick(show_args) - result = runner(show_args, cwd=cwd) - if result.returncode != 0: - return False, [], list(expected_files) - local_files = [line.strip() for line in (result.stdout or "").splitlines() if line.strip()] - equal, extra, missing = compare_effective_diff(local_files, expected_files) - # forbidden path 검사 - forbidden = detect_forbidden_paths(local_files, expected_files) - if forbidden: - return False, extra + forbidden, missing - return equal, extra, missing - - -# ─── §9 dirty tree 사전 검사 (High #2) ──────────────────────────────────── -def assert_clean_working_tree(runner: RunnerType, *, repo_dir: Optional[str] = None) -> None: - """git status --porcelain 출력이 비어있지 않으면 RuntimeError(DIRTY_WORKING_TREE).""" - cwd = repo_dir or str(WORKSPACE) - args = ["git", "status", "--porcelain"] - assert_no_forbidden_git_flags(args) - result = runner(args, cwd=cwd) - if result.returncode != 0: - raise RuntimeError(f"DIRTY_WORKING_TREE: git status failed stderr={result.stderr!r}") - if (result.stdout or "").strip(): - raise RuntimeError("DIRTY_WORKING_TREE: cannot proceed with replacement") - - -# ─── §10 replacement PR open ─────────────────────────────────────────────── -def open_replacement_pr( - task_id: str, - branch: str, - source_pr: int, - runner: RunnerType, - *, - repo_dir: Optional[str] = None, -) -> int: - """gh pr create. base=main, head=branch. body에 source PR 링크. return PR number. - repo_dir은 다른 git/gh 호출과 일관되게 worktree/temp repo에 라우팅하기 위해 필요. - """ - title = f"[{task_id}] replacement (auto for #{source_pr})" - body = f"Auto-generated replacement for #{source_pr}. Original PR preserved (no close/delete)." - args = ["gh", "pr", "create", "--base", "main", "--head", branch, "--title", title, "--body", body] - assert_no_forbidden_git_flags(args) - assert_no_cherry_pick(args) - result = runner(args, cwd=repo_dir) - if result.returncode != 0: - raise RuntimeError(f"OPEN_REPLACEMENT_PR_FAILED: stderr={result.stderr!r}") - # gh pr create는 PR URL을 stdout에 출력. 끝에서 숫자 추출. - out = (result.stdout or "").strip() - # gh pr create stdout은 https://github.com///pull/ 형식 — /pull/ 만 신뢰. - # last-digits fallback은 task id/version과 충돌 위험이 있어 제거 (Gemini high #1). - m = re.search(r"/pull/(\d+)\b", out) - if not m: - raise RuntimeError(f"OPEN_REPLACEMENT_PR_NO_PR_NUMBER: stdout={out!r}") - return int(m.group(1)) - - -# ─── §11 원 PR 보존 (close 절대 금지) ───────────────────────────────────── -def post_replaced_comment(source_pr: int, replacement_pr: int, runner: RunnerType): - """gh pr comment -b "[REPLACED] by #". close/delete 호출 금지. - args에 'close', 'delete', 'edit --state closed' 들어가면 즉시 RuntimeError. - """ - body = f"[REPLACED] by #{replacement_pr} — automated by replacement_pr_runner (task-2510). Original PR preserved." - args = ["gh", "pr", "comment", str(source_pr), "-b", body] - # 정적 검증: close/delete/edit 토큰이 args에 들어가지 않도록 - forbidden_tokens = {"close", "delete", "delete-branch"} - if any(t in args for t in forbidden_tokens): - raise RuntimeError(f"ORIGINAL_PR_PRESERVE_FORBIDDEN_OP: args={args}") - assert_no_forbidden_git_flags(args) - assert_no_cherry_pick(args) - result = runner(args) - if result.returncode != 0: - raise RuntimeError(f"POST_REPLACED_COMMENT_FAILED: stderr={result.stderr!r}") - return result - - -# ─── §12 replacement diff 사후 검증 ────────────────────────────────────── -def validate_replacement_diff( - replacement_pr: int, - expected_files: list[str], - runner: RunnerType, -) -> tuple[bool, list[str], list[str]]: - """gh pr view --json files → compare_effective_diff. - return (valid, extra, missing). - """ - args = ["gh", "pr", "view", str(replacement_pr), "--json", "files"] - result = runner(args) - if result.returncode != 0: - raise RuntimeError(f"VALIDATE_REPLACEMENT_DIFF_FAILED: pr={replacement_pr} stderr={result.stderr!r}") - data = json.loads(result.stdout or "{}") - effective = [f["path"] for f in (data.get("files") or []) if f.get("path")] - return compare_effective_diff(effective, expected_files) - - -# ─── §13 EscalationPacket helper ──────────────────────────────────────── -def build_escalation_packet( - task_id: str, - pr_number: int, - escalation_type: CriticalEscalationType, - reason: str, - evidence: dict, -) -> EscalationPacket: - """본 task에서는 EscalationPacket dataclass 인스턴스만 생성. 실제 보고는 task-2513 영역.""" - return EscalationPacket( - task_id=task_id, - pr_number=pr_number, - escalation_type=escalation_type, - reason=reason, - why_auto_cannot_continue=f"replacement_pr_runner cannot continue without chair: {reason}", - safe_options=[ - "회장 수동 검토 후 재개", - "원 PR 보존 + clean branch 수동 생성", - "expected_files 재정의", - ], - recommended_option="회장 수동 검토 후 재개", - evidence=evidence, - ) - - -# ─── §14 main runner 클래스 ──────────────────────────────────────────────── -class ReplacementPRRunner: - """contaminated PR을 받아 replacement PR을 자동 생성하는 main entry.""" - - def __init__( - self, - runner: Optional[RunnerType] = None, - *, - dry_run: bool = False, - repo_dir: Optional[str] = None, - extra_forbidden_patterns: Optional[list] = None, - timestamp_provider: Optional[Callable[[], str]] = None, - ): - self.runner = runner or _default_runner - self.dry_run = dry_run - self.repo_dir = repo_dir - self.extra_forbidden_patterns = extra_forbidden_patterns - self._ts_provider = timestamp_provider or ( - lambda: datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") - ) - self.last_escalation_packet: Optional[EscalationPacket] = None - - def _record_escalation( - self, - task_id: str, - pr_number: int, - escalation_type: CriticalEscalationType, - reason: str, - evidence: dict, - ) -> None: - """실패 경로에서 EscalationPacket을 생성하여 last_escalation_packet에 보관.""" - try: - self.last_escalation_packet = build_escalation_packet( - task_id=task_id, - pr_number=pr_number, - escalation_type=escalation_type, - reason=reason, - evidence=evidence, - ) - except Exception as e: - logger.warning("build_escalation_packet failed: %s", e) - - def execute(self, pr_number: int, task_spec: Optional[TaskSpec] = None) -> ReplacementResult: - self.last_escalation_packet = None # 매 실행마다 초기화 - - # Step 1: PR metadata - if self.dry_run: - pr_meta = self._dry_run_pr_meta(pr_number, task_spec) - else: - try: - pr_meta = fetch_pr_metadata(pr_number, self.runner) - except Exception: - task_id = (task_spec.task_id if task_spec else "unknown") or "unknown" - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, - reason="FETCH_PR_METADATA_FAILED", - evidence={"pr_number": pr_number}, - ) - # Gemini medium: 원 PR에 어떤 변경도 가하지 않았으므로 preserved=True - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=[], effective_diff_files=[], forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value, - ) - # Step 2: task_spec / expected_files - if task_spec is not None: - expected_files = list(task_spec.expected_files) - task_id = task_spec.task_id - else: - task_id = pr_meta.get("task_id") or "unknown" - expected_files = [] - # Step 3: effective diff - if self.dry_run: - effective_files = list(pr_meta.get("files", [])) - else: - try: - effective_files = compute_effective_diff(pr_meta, self.runner) - except Exception: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, - reason="COMPUTE_EFFECTIVE_DIFF_FAILED", - evidence={"pr_number": pr_number, "task_id": task_id}, - ) - # Gemini medium: 원 PR에 어떤 변경도 가하지 않았으므로 preserved=True - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=[], forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value, - ) - # Step 4: contamination - c = detect_contamination(effective_files, expected_files, extra_forbidden_patterns=self.extra_forbidden_patterns) - # Step 5: forbidden path - if c["forbidden_paths"]: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.FORBIDDEN_PATH_INTRUSION, - reason="FORBIDDEN_PATH_DETECTED", - evidence={"forbidden_paths": c["forbidden_paths"], "task_id": task_id}, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=c["forbidden_paths"], - success=False, - failure_reason=CriticalEscalationType.FORBIDDEN_PATH_INTRUSION.value, - ) - # Step 6: clean → no-op - if not c["contaminated"]: - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=True, failure_reason=None, - ) - # Step 7: contaminated → replacement - if self.dry_run: - return self._dry_run_replacement( - pr_number=pr_number, task_id=task_id, - expected_files=expected_files, effective_files=effective_files, - forbidden_paths=c["forbidden_paths"], - ) - - # 실제 모드 — High #2: pre-flight dirty tree check - try: - assert_clean_working_tree(self.runner, repo_dir=self.repo_dir) - except RuntimeError as e: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, - reason=str(e), - evidence={"dirty_tree": True, "task_id": task_id}, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value, - ) - - # High #1: commit 후 push 전 사전 검증 흐름 - branch = None - try: - branch = create_clean_replacement_branch(task_id, self.runner, timestamp=self._ts_provider(), repo_dir=self.repo_dir) - transplant_expected_files(expected_files, pr_meta["head_sha"], self.runner, repo_dir=self.repo_dir) - # commit (push X) - commit_local(task_id, self.runner, repo_dir=self.repo_dir) - except Exception as e: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, - reason=str(e), - evidence={"branch": branch, "task_id": task_id}, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value, - ) - - # ★ NEW: precheck_local_replacement_diff — push 전 사전 검증 - try: - precheck_valid, precheck_extra, precheck_missing = precheck_local_replacement_diff( - branch, expected_files, self.runner, repo_dir=self.repo_dir - ) - except Exception: - precheck_valid, precheck_extra, precheck_missing = False, [], [] - - if not precheck_valid: - # push 하지 않았으므로 local branch만 삭제 - try: - del_args = ["git", "branch", "-d", branch] - assert_no_forbidden_git_flags(del_args) - self.runner(del_args, cwd=self.repo_dir) - except Exception: - pass - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_FAILED, - reason="PRECHECK_LOCAL_DIFF_MISMATCH", - evidence={ - "branch": branch, - "extra": precheck_extra, - "missing": precheck_missing, - "task_id": task_id, - }, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value, - ) - - # 사전 검증 통과 → push + PR open - try: - push_branch(branch, self.runner, repo_dir=self.repo_dir) - replacement_pr = open_replacement_pr(task_id, branch, pr_number, self.runner, repo_dir=self.repo_dir) - except Exception as e: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF, - reason=str(e), - evidence={"branch": branch, "task_id": task_id}, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value, - ) - - # Step 8: validate replacement diff (사후 검증) - try: - valid_result = validate_replacement_diff(replacement_pr, expected_files, self.runner) - valid = valid_result if isinstance(valid_result, bool) else valid_result[0] - except Exception: - valid = False - if not valid: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_FAILED, - reason="VALIDATE_REPLACEMENT_DIFF_FAILED", - evidence={"replacement_pr": replacement_pr, "task_id": task_id}, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=replacement_pr, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value, - ) - # Step 9: 원 PR 보존 코멘트 - try: - post_replaced_comment(pr_number, replacement_pr, self.runner) - except Exception as e: - self._record_escalation( - task_id=task_id, - pr_number=pr_number, - escalation_type=CriticalEscalationType.REPLACEMENT_PR_FAILED, - reason=str(e), - evidence={"replacement_pr": replacement_pr, "task_id": task_id}, - ) - # Gemini medium: post_replaced_comment 실패해도 원 PR은 close/delete되지 않음 → preserved=True - return ReplacementResult( - source_pr=pr_number, replacement_pr=replacement_pr, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=False, - failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value, - ) - return ReplacementResult( - source_pr=pr_number, replacement_pr=replacement_pr, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=[], - success=True, failure_reason=None, - ) - - def _dry_run_pr_meta(self, pr_number, task_spec): - task_id = task_spec.task_id if task_spec else "unknown" - return { - "head_ref": f"task/{task_id}-dry-run", - "head_sha": "dry-run-sha-0000000", - "base_ref": "main", - "files": list(task_spec.expected_files) if task_spec else [], - "task_id": task_id, - "title": f"[DRY-RUN] {task_id}", - "number": pr_number, - } - - def _dry_run_replacement(self, *, pr_number, task_id, expected_files, effective_files, forbidden_paths): - timestamp = self._ts_provider() - simulated_branch = f"task/{task_id}-replacement-{timestamp}" - logger.info("[DRY-RUN] Would create branch=%r and replacement PR for PR #%d", simulated_branch, pr_number) - return ReplacementResult( - source_pr=pr_number, replacement_pr=None, original_pr_preserved=True, - expected_files=expected_files, effective_diff_files=effective_files, - forbidden_paths=forbidden_paths, - success=True, failure_reason=None, - ) - - -# ─── §15 CLI ────────────────────────────────────────────────────────────── -def main(argv: Optional[list[str]] = None) -> int: - parser = argparse.ArgumentParser(description="task-2510 replacement_pr_runner CLI") - parser.add_argument("--pr", type=int, required=True) - parser.add_argument("--dry-run", action="store_true") - parser.add_argument("--task-file", type=str, default=None) - args = parser.parse_args(argv) - runner = ReplacementPRRunner(dry_run=args.dry_run) - spec = None - if args.task_file: - spec = load_task_spec(Path(args.task_file)) - result = runner.execute(args.pr, task_spec=spec) - print(json.dumps(asdict(result), default=str, ensure_ascii=False, indent=2)) - if runner.last_escalation_packet: - print(json.dumps({"escalation_packet": asdict(runner.last_escalation_packet)}, default=str), file=sys.stderr) - return 0 if result.success else 2 - - -if __name__ == "__main__": - sys.exit(main())