"""anu_v2.post_merge_smoke_runner — ANU v2 post-merge smoke 실행 + 독립 evidence marker 박제 v0 (task-2539). 회장 §명시 (2026-05-10): - BOT squash merge 직후 main 기준 smoke 실행 - md/report fallback 절대 금지 — 실제 runner output / marker / exit code 기준으로만 판정 - smoke exit_code != 0 → Critical 7종 #7 (POST_MERGE_SMOKE_FAILURE) 분류 - one-way isolation: anu_v2 외부 import 금지 Critical 7종 매핑: #1 token raw 노출, #2 chat ID isolation 깨짐, #3 expected_files 외 수정, #4 owner_pat / admin override, #5 force / rebase, #6 manual .done 조작, #7 post-merge smoke failure ← 본 모듈 책임 """ from __future__ import annotations import json import os import re import shlex import subprocess from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Callable # ─── 모듈 레벨 상수 ──────────────────────────────────────────────────────────── DEFAULT_SMOKE_PROFILE: str = "tests/smoke/test_smoke_baseline.py" SMOKE_TIMEOUT_SECONDS: int = 300 DEFAULT_CHAT_ID: int = 6937032012 CRITICAL_SEVEN_KIND_POST_MERGE_SMOKE_FAILURE: int = 7 KIND_NAME_POST_MERGE_SMOKE_FAILURE: str = "POST_MERGE_SMOKE_FAILURE" # auto_gemini_triage 패턴 차용. import 금지이므로 모듈 내부에 재정의. # raw token 마스킹용 — lowercase 비교. TOKEN_KEY_HINTS: tuple[str, ...] = ( "github_token", "bot_github_token", "gh_token", "owner_pat", "ghp_", "ghs_", "github_pat_", "x-api-key", "authorization", "secret", "password", ) # token prefix 패턴 (값 안에서 직접 검출). 컴파일 1회. _TOKEN_PREFIX_RE = re.compile( r"(?:ghp_|ghs_|github_pat_)[A-Za-z0-9_\-]+", re.IGNORECASE, ) # KEY=value / KEY: value 패턴 마스킹용 (나머지 TOKEN_KEY_HINTS 키워드 매칭). # separator는 `=` / `:` / 공백 조합 허용. 이후 value는 공백 없는 1~200자. _KEY_VALUE_RE = re.compile( r"(?i)(" + "|".join(re.escape(h) for h in TOKEN_KEY_HINTS if h not in ("ghp_", "ghs_", "github_pat_")) + r")" r"([=:\s]+[^\s,;\"']{1,200})", ) # env 화이트리스트 _ENV_WHITELIST: frozenset[str] = frozenset({"PATH", "HOME", "LANG", "LC_ALL", "PYTHONPATH"}) # KST timezone _KST = timezone(timedelta(hours=9)) # ─── 헬퍼 ───────────────────────────────────────────────────────────────────── def _sanitize_text(text: object) -> str: """TOKEN_KEY_HINTS 관련 raw 토큰을 ***MASKED***로 치환. - ghp_ / ghs_ / github_pat_ prefix 뒤 영숫자_- → ***MASKED*** - 나머지 키워드: KEY=value / KEY:value 패턴의 value → ***MASKED*** - lowercase 비교. 너무 공격적이지 않게. """ text = text if isinstance(text, str) else str(text) # 1) prefix 직접 노출 마스킹 result = _TOKEN_PREFIX_RE.sub("***MASKED***", text) # 2) KEY=value 패턴 마스킹 result = _KEY_VALUE_RE.sub(lambda m: m.group(1) + m.group(2)[0] + "***MASKED***", result) return result def _env_whitelist() -> dict[str, str]: """subprocess env 화이트리스트 — PATH/HOME/LANG/LC_ALL/PYTHONPATH 만 통과.""" env = {} for key in _ENV_WHITELIST: val = os.environ.get(key) if val is not None: env[key] = val return env # ─── 본체 ───────────────────────────────────────────────────────────────────── class PostMergeSmokeRunner: """ANU v2 post-merge smoke 실행 + 독립 evidence marker 박제 v0. BOT squash merge 직후 main 기준 smoke 실행 후 결과를 memory/events/.smoke-evidence (jsonl) 독립 marker로 박제. 회장 §명시 (2026-05-10): - md/report fallback 절대 금지 - 실제 runner output / marker / exit code 기준으로만 판정 - smoke exit_code != 0 → Critical 7종 #7 (post-merge smoke failure) 분류 - one-way isolation: anu_v2 외부 import 금지 Critical 7종 매핑: #1 token raw 노출, #2 chat ID isolation 깨짐, #3 expected_files 외 수정, #4 owner_pat / admin override, #5 force / rebase, #6 manual .done 조작, #7 post-merge smoke failure ← 본 모듈 책임 """ DEFAULT_SMOKE_PROFILE: str = DEFAULT_SMOKE_PROFILE SMOKE_TIMEOUT_SECONDS: int = SMOKE_TIMEOUT_SECONDS def __init__( self, *, subprocess_runner: Callable[..., subprocess.CompletedProcess] | None = None, capabilities_loader: Callable[[str], dict | None] | None = None, clock: Callable[[], datetime] | None = None, workspace_root: Path | None = None, ) -> None: """주입 가능한 외부 부수효과 callable 초기화. Args: subprocess_runner: subprocess.run 대체. None이면 subprocess.run 사용. capabilities_loader: memory/capabilities/.json 읽고 smoke_command 반환. None이면 _default_capabilities_loader 사용. clock: 현재 시각 반환 callable. None이면 KST datetime.now. workspace_root: 작업 루트 디렉토리. None이면 /home/jay/workspace. """ self._subprocess_runner: Callable[..., subprocess.CompletedProcess] = ( subprocess_runner if subprocess_runner is not None else subprocess.run ) self._capabilities_loader: Callable[[str], dict | None] = ( capabilities_loader if capabilities_loader is not None else self._default_capabilities_loader ) self._clock: Callable[[], datetime] = ( clock if clock is not None else lambda: datetime.now(tz=_KST) ) self._workspace_root: Path = ( workspace_root if workspace_root is not None else Path("/home/jay/workspace") ) # ── 1. run_post_merge_smoke ──────────────────────────────────────────────── def run_post_merge_smoke( self, task_id: str, merge_commit: str, expected_files: list[str], smoke_command: str | None = None, smoke_profile: str | None = None, chat_id: int = 6937032012, ) -> dict: """BOT squash merge 직후 호출. main checkout → smoke 실행 → marker 박제. Args: task_id: e.g. "task-2537" merge_commit: full SHA40 (e.g. "da3d568aeabc29d48c9322829f36918442fa3e17") expected_files: PR의 expected_files 목록 (smoke가 이 파일들을 다루는지 검증용) smoke_command: task config 우선. None 시 smoke_profile 또는 DEFAULT_SMOKE_PROFILE smoke_profile: e.g. "tests/smoke/test_smoke_baseline.py" chat_id: 6937032012 격리 어설션 Returns: { "outcome": "PASS" | "FAIL" | "EVIDENCE_INCOMPLETE", "command": str, "exit_code": int, "duration_seconds": float, "stdout_summary": str (max 1024 chars), "stderr_summary": str (max 1024 chars), "main_head": str (resolved at run time), "merge_commit": str (input, 일치 검증), "smoke_evidence_marker_path": str | None, "critical_seven_classification": int | None # 7 if FAIL, else None } Side effects: - PASS 시 memory/events/.smoke-evidence (jsonl) 생성 내용: {task_id, merge_sha, outcome="probe_pass", ts, tests, build_ok, test_ok, command, exit_code, duration_seconds} - FAIL 시 marker 미생성, 호출자에게 critical_seven_classification=7 반환 - md/report 본문 분석 절대 금지 (회장 §명시) """ # ── chat isolation 어설션 ────────────────────────────────────────────── assert chat_id == DEFAULT_CHAT_ID, "chat_id != 6937032012 (cross-chat 누출 차단)" # ── 입력 검증 ────────────────────────────────────────────────────────── assert task_id, "task_id must not be empty" if len(merge_commit) != 40: raise ValueError( f"merge_commit must be full SHA40 (got length {len(merge_commit)}): {merge_commit!r}" ) # ── main_head resolve ────────────────────────────────────────────────── main_head: str | None = None try: git_result = self._subprocess_runner( ["git", "rev-parse", "origin/main"], capture_output=True, text=True, check=False, cwd=str(self._workspace_root), ) if git_result.returncode == 0: main_head = git_result.stdout.strip() except Exception: main_head = None if not main_head: command = self._resolve_smoke_command(task_id, smoke_command, smoke_profile) return { "outcome": "EVIDENCE_INCOMPLETE", "command": command, "exit_code": None, "duration_seconds": None, "stdout_summary": "", "stderr_summary": "", "main_head": None, "merge_commit": merge_commit, "smoke_evidence_marker_path": None, "critical_seven_classification": None, } # ── smoke command 결정 ──────────────────────────────────────────────── command = self._resolve_smoke_command(task_id, smoke_command, smoke_profile) # ── smoke 실행 ──────────────────────────────────────────────────────── execution_result = self._execute_smoke(command) exit_code: int = execution_result["exit_code"] duration: float = execution_result["duration_seconds"] stdout_raw: str = execution_result["stdout"] stderr_raw: str = execution_result["stderr"] stdout_summary = _sanitize_text(stdout_raw)[:1024] stderr_summary = _sanitize_text(stderr_raw)[:1024] # ── 결과 판정 ────────────────────────────────────────────────────────── if exit_code == 0: # PASS 경로 evidence = self._build_smoke_evidence( task_id, merge_commit, execution_result, main_head, expected_files_count=len(expected_files), ) marker_path: str | None = None try: marker_path = self._write_smoke_evidence_marker(task_id, evidence) except OSError: return { "outcome": "EVIDENCE_INCOMPLETE", "command": _sanitize_text(command), "exit_code": exit_code, "duration_seconds": duration, "stdout_summary": stdout_summary, "stderr_summary": stderr_summary, "main_head": main_head, "merge_commit": merge_commit, "smoke_evidence_marker_path": None, "critical_seven_classification": None, } # ── PASS 직후 worktree cleanup dry-run (task-2550+1 통합) ──────────── # smoke PASS 결과에는 영향 X — try/except 로 모든 예외 흡수 cleanup_summary: dict | None = None cleanup_error: str | None = None try: cleanup_summary = self.run_post_merge_worktree_cleanup_dry_run(task_id, apply=False) except Exception as e: # noqa: BLE001 — cleanup 실패는 smoke PASS 결과에 영향 X (operational nudge), 단 에러는 보고 cleanup_error = _sanitize_text(str(e))[:200] return { "outcome": "PASS", "command": _sanitize_text(command), "exit_code": exit_code, "duration_seconds": duration, "stdout_summary": stdout_summary, "stderr_summary": stderr_summary, "main_head": main_head, "merge_commit": merge_commit, "smoke_evidence_marker_path": marker_path, "critical_seven_classification": None, "worktree_cleanup_summary": cleanup_summary, "worktree_cleanup_error": cleanup_error, } else: # FAIL 경로 → Critical 7종 #7 return { "outcome": "FAIL", "command": _sanitize_text(command), "exit_code": exit_code, "duration_seconds": duration, "stdout_summary": stdout_summary, "stderr_summary": stderr_summary, "main_head": main_head, "merge_commit": merge_commit, "smoke_evidence_marker_path": None, "critical_seven_classification": CRITICAL_SEVEN_KIND_POST_MERGE_SMOKE_FAILURE, } # ── 2. _resolve_smoke_command ───────────────────────────────────────────── def _resolve_smoke_command( self, task_id: str, smoke_command: str | None, smoke_profile: str | None, ) -> str: """smoke command 결정 우선순위: 1. 명시 smoke_command 입력 2. task config (memory/capabilities/.json::smoke_command) 3. smoke_profile 입력 → f"python3 -m pytest {smoke_profile}" 래핑 4. f"python3 -m pytest {DEFAULT_SMOKE_PROFILE}" """ # 1. 명시 입력 if smoke_command: return smoke_command # 2. capabilities loader caps = self._capabilities_loader(task_id) if caps and caps.get("smoke_command"): return caps["smoke_command"] # 3. smoke_profile 입력 if smoke_profile: return f"python3 -m pytest {smoke_profile}" # 4. DEFAULT return f"python3 -m pytest {DEFAULT_SMOKE_PROFILE}" # ── 3. _execute_smoke ───────────────────────────────────────────────────── def _execute_smoke(self, command: str, timeout: int = SMOKE_TIMEOUT_SECONDS) -> dict: """subprocess.run으로 실행. timeout / exit_code / stdout / stderr / duration 반환. - check=False (FAIL 시 Critical 7종 분류 위해 raise 안 함) - text=True - cwd=workspace root - env에서 token/key 노출 0 어설션 (env 화이트리스트만 통과) """ t_start = self._clock() try: proc = self._subprocess_runner( shlex.split(command), capture_output=True, text=True, check=False, cwd=str(self._workspace_root), timeout=timeout, env=_env_whitelist(), ) t_end = self._clock() duration = (t_end - t_start).total_seconds() return { "command": command, "exit_code": proc.returncode, "stdout": proc.stdout or "", "stderr": proc.stderr or "", "duration_seconds": duration, } except subprocess.TimeoutExpired: return { "command": command, "exit_code": 124, "stdout": "", "stderr": "", "duration_seconds": float(timeout), } # ── 4. _build_smoke_evidence ────────────────────────────────────────────── def _build_smoke_evidence( self, task_id: str, merge_commit: str, execution_result: dict, main_head: str, expected_files_count: int = 0, ) -> dict: """marker 본문 dict 생성. task-2524 박제 형식 1:1 호환. format: { "task_id": "task-XXXX", "merge_sha": "", "outcome": "probe_pass", "ts": "", "tests": " exit=", "build_ok": bool, "test_ok": bool, "command": str, "exit_code": int, "duration_seconds": float, "main_head": str, "merge_commit": str, "expected_files_count": int, # smoke가 다뤄야 할 expected_files 수 (검증 hook) } 토큰 raw 0 / chat_id 0 / API key 0 (회장 §명시 박제 원칙) """ command = execution_result["command"] exit_code = execution_result["exit_code"] duration = execution_result["duration_seconds"] return { "task_id": _sanitize_text(task_id), "merge_sha": merge_commit, # 40자 full SHA — sanitize 불필요 "outcome": "probe_pass", # task-2524 박제 형식 따름 "ts": self._clock().isoformat(), "tests": _sanitize_text(f"{command} exit={exit_code}"), "build_ok": True, "test_ok": True, "command": _sanitize_text(command), "exit_code": exit_code, "duration_seconds": duration, "main_head": _sanitize_text(main_head), "merge_commit": merge_commit, "expected_files_count": expected_files_count, } # ── 5. _write_smoke_evidence_marker ────────────────────────────────────── def _write_smoke_evidence_marker( self, task_id: str, evidence: dict, marker_dir: str = "memory/events", ) -> str: """memory/events/.smoke-evidence 작성 (jsonl). idempotent: 기존 marker 존재 시 append (덮어쓰기 X). chat_id != 6937032012 record 절대 작성 X. Args: task_id: e.g. "task-2539" evidence: _build_smoke_evidence 반환값 marker_dir: workspace_root 기준 상대 경로 Returns: marker 절대 경로 str """ # evidence dict에 chat_id가 있으면 격리 검증 if "chat_id" in evidence: assert evidence["chat_id"] == DEFAULT_CHAT_ID, ( "chat_id != 6937032012 — cross-chat record 작성 차단" ) marker_path = self._workspace_root / marker_dir / f"{task_id}.smoke-evidence" marker_path.parent.mkdir(parents=True, exist_ok=True) # jsonl append (idempotent — 기존 marker 존재 시 line 추가, 덮어쓰기 X) with open(marker_path, "a", encoding="utf-8") as f: f.write(json.dumps(evidence, ensure_ascii=False) + "\n") return str(marker_path.resolve()) # ── 6. classify_smoke_failure_as_critical_seven ─────────────────────────── def classify_smoke_failure_as_critical_seven( self, task_id: str, merge_commit: str, execution_result: dict, expected_files_count: int = 0, ) -> dict: """smoke exit_code != 0 시 Critical 7종 #7 분류 박제. Returns: { "critical_seven_kind": 7, "kind_name": "POST_MERGE_SMOKE_FAILURE", "task_id": str, "merge_commit": str, "exit_code": int, "command": str, "stderr_summary": str (max 512 chars, token 0 어설션), "ts": str, "report_to_chairman_required": True, "expected_files_count": int, } 본 메서드 자체는 회장 보고 X (반환만). 실제 회장 보고는 호출자 책임. """ return { "critical_seven_kind": CRITICAL_SEVEN_KIND_POST_MERGE_SMOKE_FAILURE, "kind_name": KIND_NAME_POST_MERGE_SMOKE_FAILURE, "task_id": task_id, "merge_commit": merge_commit, "exit_code": execution_result["exit_code"], "command": _sanitize_text(execution_result["command"]), "stderr_summary": _sanitize_text(execution_result["stderr"])[:512], "ts": self._clock().isoformat(), "report_to_chairman_required": True, "expected_files_count": expected_files_count, } # ── 내부 헬퍼 ───────────────────────────────────────────────────────────── def _default_capabilities_loader(self, task_id: str) -> dict | None: """memory/capabilities/.json 읽고 smoke_command 키 반환. 파일이 없거나 읽기 실패 시 None 반환. """ caps_path = self._workspace_root / "memory" / "capabilities" / f"{task_id}.json" try: with open(caps_path, encoding="utf-8") as f: return json.load(f) except (OSError, json.JSONDecodeError): return None # ── 7. run_post_merge_worktree_cleanup_dry_run (task-2550+1 통합) ───────── def run_post_merge_worktree_cleanup_dry_run( self, task_id: str, apply: bool = False, ) -> dict: """post-merge 후 worktree cleanup dry-run 1회 시도. Args: task_id: 현재 머지된 task_id (참고용 — cleanup은 전체 worktree 대상) apply: 실제 삭제 여부. 기본 False (dry-run). True 시 회장 별도 승인 의미. Returns: { "task_id": str, "apply": bool, "total_worktrees": int, "cleanup_candidates": int, # safety 1~5 PASS + not main + not dirty (apply_explicit 제외) "applied_count": int, # 실제 삭제된 수 "skipped_count": int, "dirty_skipped": int, "main_protected": int, "ts": str, "results": list[dict], # 각 worktree CleanupResult.asdict() } Side effects: - dirty worktree skip 시 memory/events/worktree-cleanup-skipped--.json 작성 - apply=True 시 git worktree remove 실행 task-2550+1 medium fix (cleanup_candidates 산정): - 기존 `r.all_safe` 는 dry-run 에서 항상 False (apply_explicit FAIL) → cleanup_candidates 항상 0 으로 가시성 상실. - 신규: `is_safe_ignoring_apply(r)` helper 사용 — safety 1~5 PASS + not main + not dirty 로 산정 (apply_explicit 단독 FAIL 은 dry-run 정상 동작이므로 candidate 카운트에서 분리). """ # anu_v2.worktree_cleanup import (anu_v2 내부 import, isolation 보장) # pyright: ignore[reportMissingImports] — worktree 환경에서 main workspace extraPaths가 새 파일을 보지 못함. 머지 후 해결. from anu_v2.worktree_cleanup import WorktreeCleanup, is_safe_ignoring_apply # pyright: ignore[reportMissingImports] cleanup = WorktreeCleanup( subprocess_runner=self._subprocess_runner, clock=self._clock, workspace_root=self._workspace_root, ) results = cleanup.cleanup_all_dry_run(apply=apply) # task-2550+1 medium fix: apply_explicit 제외하고 candidate 산정 (가시성 회복) cleanup_candidates = sum(1 for r in results if is_safe_ignoring_apply(r)) applied_count = sum(1 for r in results if r.applied) skipped_count = sum(1 for r in results if r.skipped) dirty_skipped = sum(1 for r in results if r.dirty) main_protected = sum(1 for r in results if r.is_main) from dataclasses import asdict return { "task_id": task_id, "apply": apply, "total_worktrees": len(results), "cleanup_candidates": cleanup_candidates, "applied_count": applied_count, "skipped_count": skipped_count, "dirty_skipped": dirty_skipped, "main_protected": main_protected, "ts": self._clock().isoformat(), "results": [asdict(r) for r in results], }