"""task-2556 §4 — follow-up commit 후 GEMINI_STALE_ON_HEAD 자동 trigger 회귀. 회장 §명시 2026-05-12 §4: PR head SHA != Gemini review commit_id → GEMINI_STALE_ON_HEAD → owner trigger 필요. 검증 포인트: 1. head_sha 와 latest review commit_id 가 다르면 STALE. 2. scheduler 가 OWNER_TRIGGER_DISPATCHED. 3. decision.json 의 current_head 가 새 head_sha (review commit_id 가 아님). 4. fresh review 가 head 와 일치하면 GEMINI_FRESH_ON_HEAD → SKIP. 5. 여러 reviews 중 마지막 (최신) 만 비교 대상. 6. fixture (gemini_stale_on_head_2556.json) 결과 어셀션. """ from __future__ import annotations import json import sys from pathlib import Path WORKSPACE_ROOT = Path(__file__).resolve().parents[2] if str(WORKSPACE_ROOT) not in sys.path: sys.path.insert(0, str(WORKSPACE_ROOT)) from anu_v2.executor_scheduler import ( # noqa: E402 ACTION_FRESH_RESUME, ACTION_OWNER_TRIGGER_DISPATCHED, ExecutorScheduler, ) from anu_v2.idle_pr_diagnoser import ( # noqa: E402 GeminiReviewMeta, IdlePRDiagnoser, IdlePRSnapshot, STATE_GEMINI_FRESH_ON_HEAD, STATE_GEMINI_STALE_ON_HEAD, ) from anu_v2.merge_queue_executor import MergeQueueExecutor # noqa: E402 from anu_v2.owner_trigger_audit import OwnerTriggerAudit # noqa: E402 from anu_v2.owner_trigger_only import OwnerTriggerOnly # noqa: E402 _HEAD_NEW = "b" * 40 _HEAD_OLD = "a" * 40 def _snapshot(reviews=(), head=_HEAD_NEW, number=108): return IdlePRSnapshot( number=number, head_sha=head, head_ref="task/task-2556-dev5", created_at="2026-05-12T10:00:00+00:00", gemini_reviews=tuple(reviews), ci_required_all_success=True, ) def _make_runner(tmp_path: Path, *, http_calls: list, token: str = "ghp_token_fake_xxxxxx"): def http_post(method, path, body, headers): http_calls.append({"method": method, "path": path, "body": body}) return {"id": 1} return OwnerTriggerOnly( workspace_root=tmp_path, http_post=http_post, token_provider=lambda: token, audit=OwnerTriggerAudit(tmp_path), ) def _make_scheduler(tmp_path: Path, snapshots, *, http_calls: list): return ExecutorScheduler( workspace_root=tmp_path, decision_dir=tmp_path / "memory" / "events", snapshot_provider=lambda: snapshots, owner_trigger=_make_runner(tmp_path, http_calls=http_calls), merge_executor=MergeQueueExecutor( gh_runner=lambda a, e: {}, git_runner=lambda a: "", pytest_runner=lambda p: 0, audit_writer=lambda p: None, task_md_root=tmp_path, ), owner="o", repo="r", ) def test_stale_review_classified_gemini_stale_on_head(): """latest review commit_id != head_sha → STALE.""" review = GeminiReviewMeta(commit_id=_HEAD_OLD, submitted_at="2026-05-12T10:15:00+00:00") diag = IdlePRDiagnoser().diagnose( _snapshot(reviews=(review,)), now="2026-05-12T12:00:00+00:00", ) assert diag.state == STATE_GEMINI_STALE_ON_HEAD assert diag.head_sha == _HEAD_NEW assert diag.latest_gemini_commit_id == _HEAD_OLD assert diag.requires_owner_trigger is True def test_scheduler_dispatches_owner_trigger_for_stale_pr(tmp_path): http_calls: list = [] review = GeminiReviewMeta(commit_id=_HEAD_OLD, submitted_at="2026-05-12T10:15:00+00:00") snap = _snapshot(reviews=(review,), number=108) scheduler = _make_scheduler(tmp_path, [snap], http_calls=http_calls) result = scheduler.run_one_cycle( env={"OWNER_GEMINI_TRIGGER_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxxxxxx"}, now="2026-05-12T12:30:00+00:00", ) action = result.pr_actions[0] assert action.state == STATE_GEMINI_STALE_ON_HEAD assert action.action == ACTION_OWNER_TRIGGER_DISPATCHED decision_path = tmp_path / "memory" / "events" / "task-2556.owner_trigger_decision.json" decision = json.loads(decision_path.read_text(encoding="utf-8")) # decision.current_head = new head, NOT old commit_id (회장 §4 박제) assert decision["current_head"] == _HEAD_NEW assert decision["gemini_evidence_fresh"] is False assert len(http_calls) == 1 assert http_calls[0]["path"] == "/repos/o/r/issues/108/comments" def test_multiple_reviews_uses_latest_only(): """여러 review 중 마지막 (최신) 만 비교 대상이어야 함.""" older = GeminiReviewMeta(commit_id="c" * 40, submitted_at="2026-05-12T10:10:00+00:00") newer = GeminiReviewMeta(commit_id=_HEAD_OLD, submitted_at="2026-05-12T10:20:00+00:00") diag = IdlePRDiagnoser().diagnose( _snapshot(reviews=(older, newer)), now="2026-05-12T12:00:00+00:00", ) assert diag.state == STATE_GEMINI_STALE_ON_HEAD assert diag.latest_gemini_commit_id == _HEAD_OLD def test_fresh_review_classified_gemini_fresh_on_head(tmp_path): review = GeminiReviewMeta(commit_id=_HEAD_NEW, submitted_at="2026-05-12T11:00:00+00:00") snap = _snapshot(reviews=(review,)) diag = IdlePRDiagnoser().diagnose(snap, now="2026-05-12T12:00:00+00:00") assert diag.state == STATE_GEMINI_FRESH_ON_HEAD assert diag.requires_owner_trigger is False http_calls: list = [] scheduler = _make_scheduler(tmp_path, [snap], http_calls=http_calls) result = scheduler.run_one_cycle( env={"OWNER_GEMINI_TRIGGER_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxxxxxx"}, now="2026-05-12T12:00:00+00:00", ) assert result.pr_actions[0].action == ACTION_FRESH_RESUME assert len(http_calls) == 0 def test_stale_fixture_matches_diagnoser_output(): fixture_path = ( Path(__file__).resolve().parents[1] / "fixtures" / "gemini_stale_on_head_2556.json" ) fixture = json.loads(fixture_path.read_text(encoding="utf-8")) snap_dict = fixture["snapshot"] reviews = tuple( GeminiReviewMeta(commit_id=r["commit_id"], submitted_at=r["submitted_at"]) for r in snap_dict["gemini_reviews"] ) snap = IdlePRSnapshot( number=snap_dict["number"], head_sha=snap_dict["head_sha"], head_ref=snap_dict["head_ref"], created_at=snap_dict["created_at"], gemini_reviews=reviews, ci_required_all_success=snap_dict["ci_required_all_success"], ) diag = IdlePRDiagnoser().diagnose(snap, now=fixture["now"]) assert diag.state == fixture["expected_diagnosis"]["state"] assert diag.head_sha == fixture["expected_diagnosis"]["head_sha"] assert ( diag.latest_gemini_commit_id == fixture["expected_diagnosis"]["latest_gemini_commit_id"] )