#!/usr/bin/env python3 """run_post_merge_reconcile_closeout.py — task-2553+13 Track A orchestrator (fail-closed). 회장 병행 GO — Track A. task-2553+12 = PR #128 BOT merge SUCCESS(mergeCommit d08b8b0efa4d44fea99f1e5e391c1a18695e85f6, 비가역) + post-merge smoke false-negative → POST_MERGE_HOLD. 본 runner = harness-artifact 분류 확정 + reconcile evidence + task-2553+1 closeout 를 코드/파일 자동화로 마감. md-only 완료처리 금지 — 모든 산출 JSON/evidence + evidence-gated .done. 흐름(fail-closed): preflight(live ws invariant capture) → classify(독립 격리 재현 + §1 근거) → reconcile(9-R.2 단일 observation window, contains-not-tip, read-only) → closeout evidence 8-key 독립 재검증(9-R.4) → (전건 PASS only) schema-safe closeout(9-R.3: sidecar 단일권위 + result.json 최상위 key 비파괴 추가 + md EOF append) → evidence-gated .done → final packet → postflight (live ws invariant assertEqual). 안전/금지(task-2553+13.md §5 / 9-R): - code change 0: 기존 production·test·operational code byte 변경 0. 본 runner 신규 생성 = Track A automation deliverable(9-R.1). - git/gh = 전부 read-only(fetch/rev-parse/merge-base/rev-list/diff/show/ gh pr view/gh api). checkout·reset·stash·rm·commit·push·merge 정적 부재. - 9-R.2 reconcile = origin/main 이 mergeCommit '포함'(is-ancestor, tip== 금지, race-free) + 단일 fetch observation window(ref SHA+ts 기록). - 9-R.3 task-2553+1.result.json in-place 문자열 append 금지 — read→parse→ 최상위 신규 key 비파괴 추가→atomic write. 파싱 실패·기존 key 변형 → HOLD. sidecar memory/events/task-2553+1.closeout.json = 단일 권위. - 9-R.4 .done = closeout.json 모든 완료기준 키를 runner 가 독립 재검증한 뒤에만 생성. evidence_consistent boolean 단독 신뢰 금지. 1개라도 미증명 → .done 0 + HOLD. - 9-R.5 namespace = task-2553+13.* / task-2553+1.closeout* read/write 한정. 병행 Track B artifact(타 트랙 산출 경로) glob·open·인용·근거 사용 0 (정적 부재 — 본 파일은 해당 경로를 일절 참조하지 않음). - HOLD 적중 시 closeout/.done 생성 0 + task-2553+13.hold-for-chair.json. - live /home/jay/workspace @ task/task-2553p1-f1-clean-replacement 20456b5f 전후 HEAD/branch assertEqual. cleanup·reset·stash·rm 0. """ from __future__ import annotations import hashlib import json import os import subprocess import sys import tarfile import tempfile import time import importlib.util from datetime import datetime, timezone from pathlib import Path # ─── 불변 상수 (task-2553+12 확정 입력 / task-2553+13.md §1) ────────────────── REPO = "Jeon-Jonghyuk/dev_workspace" PR = 128 PR102 = 102 MERGE_COMMIT = "d08b8b0efa4d44fea99f1e5e391c1a18695e85f6" MERGED_AT = "2026-05-17T06:12:13Z" MERGED_BY = "app/jeon-jonghyuk-taskctl-bot" PINNED_HEAD = "0ea36fc9a724b1763be34710e283e088fae39a59" # parent2 (PR#128 head) BASE_PARENT = "7346df8260803308df30a6d04ec32d66d4cdfa5b" # parent1 (fresh base) PR102_HEAD = "bd5ad74f5d443b354319fc8b3cb006816b8a9025" # F2 token-transport 블록 = task-2553 계열 정준 정의(owner_trigger_pat.py # lines 119-156). task-2553+12.pre-merge-gate / run_pr128_merge_closeout.py # 와 동일 정의·sha 사용(lineage 정합). PR#102==merged 동일이어야 byte-identical. F2_LINES = (119, 156) F2_REGION_SHA_EXPECT = "b02140738e372578a8f39af3d8ca3e13ce8ec099f86393a49e1e224a3f6a7560" MERGED_OTP_FILE_SHA = "7b7d996aae3c368561f63600f8e71017f7af85b86a63b5533153e956bdec7135" EXPECTED_6 = sorted([ "anu_v2/owner_trigger_pat.py", "memory/events/task-2553+1.green-evidence.log", "memory/events/task-2553+1.red-evidence.log", "memory/events/task-2553+1.result.json", "memory/reports/task-2553+1.md", "tests/regression/test_owner_trigger_2553_plus1_high_fix.py", ]) # CI 11 = 고유 10 + 1 중복(taskctl-state-guard 2회 run). 전건 SUCCESS 필요. CI_REQUIRED_UNIQUE = sorted([ "cancel-kill-switch", "ci/guard", "gemini-review-gate", "guard", "hidden-path-audit", "lock-in-check", "merge-safety-check", "phase3-merge-gate", "qc-check", "taskctl-state-guard", ]) CI_TOTAL_EXPECT = 11 # 보존(preserved) 테스트 = effective diff 에 신규 1파일만 허용, 아래는 미포함이어야 함. PRESERVED_TEST_HINTS = ( "test_owner_trigger_success_2553", "test_owner_trigger_security_boundaries_2553", "test_owner_trigger_pat_phase2_2553", "test_owner_trigger_pat_phase3_integration_2553", ) NEW_TEST_FILE = "tests/regression/test_owner_trigger_2553_plus1_high_fix.py" LIVE_WS = Path("/home/jay/workspace") LIVE_WS_EXPECT_BRANCH = "task/task-2553p1-f1-clean-replacement" LIVE_WS_EXPECT_HEAD = "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" TASK = "task-2553+13" TEAM = "dev5-team" EV = LIVE_WS / "memory" / "events" RP = LIVE_WS / "memory" / "reports" def _utc() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _git(*args: str) -> subprocess.CompletedProcess: """read-only git only. 변이성 subcommand 정적 차단.""" banned = {"checkout", "reset", "stash", "rm", "commit", "push", "merge", "rebase", "clean", "switch", "restore", "branch", "tag", "apply"} if args and args[0] in banned: raise RuntimeError(f"FORBIDDEN git mutation blocked: git {args[0]}") return subprocess.run(["git", "-C", str(LIVE_WS), *args], capture_output=True, text=True) def _gh_json(*args: str): cp = subprocess.run(["gh", *args], capture_output=True, text=True, cwd=str(LIVE_WS)) if cp.returncode != 0: return None, cp.stderr.strip() try: return json.loads(cp.stdout), None except json.JSONDecodeError as e: return None, f"json decode: {e}" def _sha256_bytes(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def _atomic_write(path: Path, data: str) -> None: tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(data, encoding="utf-8") os.replace(tmp, path) def _write_json(path: Path, obj) -> None: _atomic_write(path, json.dumps(obj, ensure_ascii=False, indent=2) + "\n") class Hold(Exception): def __init__(self, reason: str, detail): super().__init__(reason) self.reason = reason self.detail = detail def _emit_hold(reason: str, detail, started: str) -> None: """HOLD: closeout/.done 생성 0. hold-for-chair.json 만 기록.""" _write_json(EV / f"{TASK}.hold-for-chair.json", { "task": TASK, "ts": _utc(), "started": started, "status": "HOLD_FOR_CHAIR", "stage": "TRACK_A", "reason": reason, "detail": detail, "closeout_created": False, "done_created": False, "guarantee": "evidence 부정합/무결성 미확인 — closeout·.done 생성 0", "live_workspace_invariant": _ws_state(), }) def _ws_state() -> dict: head = _git("rev-parse", "HEAD").stdout.strip() branch = _git("rev-parse", "--abbrev-ref", "HEAD").stdout.strip() return {"head": head, "branch": branch} # ─── 1. classify — harness-artifact (독립 격리 재현 + §1 근거) ──────────────── def classify() -> dict: """merge SUCCESS + post-merge smoke harness false-negative. 격리 tmp 에 mergeCommit owner_trigger_pat.py 추출 후 2-branch 결정적 재현: A) sys.modules 미등록(=test _load_otp 설계) → py3.12 @dataclass AttributeError B) sys.modules 등록(정상 로더) → IMPORT_OK True + 운영 심볼 정상 live workspace·운영코드 미접촉(read-only git archive → /tmp). """ repro = {} cwd0 = os.getcwd() tmp = tempfile.mkdtemp(prefix="pmrc_repro_") try: ar = subprocess.run( ["git", "-C", str(LIVE_WS), "archive", MERGE_COMMIT, "anu_v2/owner_trigger_pat.py"], capture_output=True) if ar.returncode != 0: raise Hold("repro_archive_failed", ar.stderr.decode()[:300]) with tarfile.open(fileobj=__import__("io").BytesIO(ar.stdout)) as tf: tf.extractall(tmp) otp = os.path.join(tmp, "anu_v2", "owner_trigger_pat.py") repro["target_file_sha256"] = _sha256_bytes(Path(otp).read_bytes()) # branch A: 미등록 spec = importlib.util.spec_from_file_location("otp_noreg_a", otp) if spec is None or spec.loader is None: raise Hold("repro_spec_none", "branch A spec/loader None") m = importlib.util.module_from_spec(spec) try: spec.loader.exec_module(m) a_ok, a_err = True, None except Exception as e: # noqa: BLE001 a_ok, a_err = False, f"{type(e).__name__}: {e}" # branch B: 등록(정상 로더) spec2 = importlib.util.spec_from_file_location("otp_reg_b", otp) if spec2 is None or spec2.loader is None: raise Hold("repro_spec_none", "branch B spec/loader None") m2 = importlib.util.module_from_spec(spec2) sys.modules["otp_reg_b"] = m2 try: spec2.loader.exec_module(m2) b_ok, b_err = True, None except Exception as e: # noqa: BLE001 b_ok, b_err = False, f"{type(e).__name__}: {e}" finally: sys.modules.pop("otp_reg_b", None) repro["branch_A_no_sysmodules"] = { "loader": "spec_from_file_location, sys.modules[name] 미등록 (= test " "_load_otp 헬퍼 설계)", "import_ok": a_ok, "error": a_err, "expect": "AttributeError (py3.12 dataclasses.py:749 sys.modules." "get(cls.__module__).__dict__ → None)", } repro["branch_B_with_sysmodules"] = { "loader": "spec_from_file_location + sys.modules[name]=mod (정상 로더)", "import_ok": b_ok, "error": b_err, "has_is_duplicate_trigger": (hasattr(m2, "is_duplicate_trigger") if b_ok else None), "ALLOWED_COMMENT_BODY": (getattr(m2, "ALLOWED_COMMENT_BODY", None) if b_ok else None), } finally: os.chdir(cwd0) subprocess.run(["rm", "-rf", tmp]) # /tmp 격리 디렉터리만 — live 무관 a = repro["branch_A_no_sysmodules"] b = repro["branch_B_with_sysmodules"] mechanism_proven = ( a["import_ok"] is False and "AttributeError" in (a["error"] or "") and b["import_ok"] is True and b["has_is_duplicate_trigger"] is True and b["ALLOWED_COMMENT_BODY"] == "/gemini review" ) decision = ("MERGE_SUCCESS + POST_MERGE_SMOKE_HARNESS_FALSE_NEGATIVE" if mechanism_proven else "INDETERMINATE") obj = { "task": TASK, "ts": _utc(), "classification": decision, "merge_success_irreversible": { "merged": True, "mergeCommit": MERGE_COMMIT, "mergedBy": MERGED_BY, "mergedAt": MERGED_AT, "method": "merge", "pinned_head": PINNED_HEAD, "rollback_revert_force": "NOT_PERFORMED (금지·비가역 유지)", }, "smoke_failure_is_harness_artifact": True, "root_cause": { "type": "TEST/HARNESS LOADER ARTIFACT — NOT merged-code regression", "mechanism": "test 의 _load_otp() 가 importlib spec_from_file_location " "exec 시 sys.modules[name]=mod 등록 생략 → Python 3.12 " "owner_trigger_pat.py @dataclass 처리(dataclasses.py:749 " "sys.modules.get(cls.__module__).__dict__) None 참조 → " "AttributeError.", "scope": "3 F1 케이스(_load_otp 경유) FAIL / 스트리밍 4 케이스" "(_load_otp_streaming 경유) PASS.", "failing_tests": [ "test_allowed_comment_body_is_exactly_gemini_review", "test_args_allowlist_rejects_foreign_endpoint", "test_args_allowlist_rejects_foreign_body", ], }, "grounds": { "ci_11_success": "PR #128 CI 11/11 SUCCESS (qc-check·" "gemini-review-gate 포함) — gh statusCheckRollup 재조회", "independent_reproduction": "격리 tarball(d08b8b0e) 2-branch 결정적 " "재현 — 본 JSON repro 필드", "loader_mechanism": "sys.modules 미등록 메커니즘 입증(branch A)", "failing_3_vs_streaming_4": "F1 3 FAIL / streaming 4 PASS 비대칭이 " "로더 헬퍼 경유 차이와 정확히 일치", }, "deterministic_reproduction": { "command": "git archive d08b8b0e anu_v2/owner_trigger_pat.py → " "/tmp 격리 → spec_from_file_location 2-branch import", "result": repro, "live_workspace_touched": False, "production_code_touched": False, }, "mechanism_proven": mechanism_proven, "merged_code_health": ("CONFIRMED — IMPORT_OK True, is_duplicate_trigger " "존재, ALLOWED_COMMENT_BODY=='/gemini review'"), } _write_json(EV / f"{TASK}.harness-artifact-classification.json", obj) if not mechanism_proven: raise Hold("classification_grounds_insufficient", {"repro": repro}) return obj # ─── 2. reconcile — 9-R.2 단일 observation window, read-only ────────────────── def reconcile() -> dict: obs_start = _utc() f = _git("fetch", "origin", "main") # read-only: remote-tracking only if f.returncode != 0: raise Hold("git_fetch_failed", f.stderr.strip()[:300]) # MED-fix: 단일 observation window = 캡처한 1개 ref SHA 에 모든 containment # 검사를 고정(symbolic origin/main 재해석 금지 — strict single-SHA window). origin_main = _git("rev-parse", "origin/main").stdout.strip() fetch_head = _git("rev-parse", "FETCH_HEAD").stdout.strip() obs_end = _utc() if not origin_main or origin_main != fetch_head: raise Hold("observation_window_unstable", {"origin_main": origin_main, "fetch_head": fetch_head}) PIN = origin_main # 이후 모든 검사는 이 SHA 에만 고정 # ① contains mergeCommit (PIN 기준 is-ancestor; tip== 금지·race-free) anc = _git("merge-base", "--is-ancestor", MERGE_COMMIT, PIN) is_ancestor = anc.returncode == 0 rl = _git("rev-list", PIN) contains_revlist = MERGE_COMMIT in rl.stdout.split() contains = is_ancestor and contains_revlist # ② effective diff parent1..mergeCommit == EXPECTED_6 df = _git("diff", "--name-only", BASE_PARENT, MERGE_COMMIT) actual_6 = sorted(x for x in df.stdout.splitlines() if x.strip()) diff_ok = actual_6 == EXPECTED_6 # ③ gh pr view 128 pr128, e1 = _gh_json("pr", "view", str(PR), "--json", "number,state,mergeCommit,mergedAt,mergedBy," "baseRefName,headRefName,headRefOid") pr128_ok = bool(pr128) and ( pr128.get("state") == "MERGED" and (pr128.get("mergeCommit") or {}).get("oid") == MERGE_COMMIT and pr128.get("mergedAt") == MERGED_AT and pr128.get("headRefOid") == PINNED_HEAD) # ④ gh pr view 102 (보존: OPEN·head 무변) pr102, e2 = _gh_json("pr", "view", str(PR102), "--json", "number,state,headRefOid,headRefName,baseRefName") pr102_ok = bool(pr102) and ( pr102.get("state") == "OPEN" and pr102.get("headRefOid") == PR102_HEAD) reconcile_pass = contains and diff_ok and pr128_ok and pr102_ok obj = { "task": TASK, "ts": _utc(), "method": "9-R.2 single observation window, read-only fetch", "observation_window": { "start_utc": obs_start, "end_utc": obs_end, "git_fetch": "origin main (remote-tracking only — read-only)", "pinned_ref_sha": PIN, "fetched_origin_main_sha": origin_main, "fetched_FETCH_HEAD_sha": fetch_head, "single_sha_window": True, "note": "모든 containment 검사를 캡처 SHA(PIN)에 고정 — symbolic " "origin/main 재해석 0 (TOCTOU 최소화).", }, "check_1_contains_mergeCommit": { "semantic": "origin/main(PIN) 이 mergeCommit 포함 (tip== 금지·" "race-free)", "merge_base_is_ancestor": is_ancestor, "rev_list_contains": contains_revlist, "pass": contains, }, "check_2_effective_diff_6": { "range": f"{BASE_PARENT[:8]}..{MERGE_COMMIT[:8]}", "expected": EXPECTED_6, "actual": actual_6, "pass": diff_ok, }, "check_3_pr128_merged": { "state": (pr128 or {}).get("state"), "mergeCommit": ((pr128 or {}).get("mergeCommit") or {}).get("oid"), "mergedAt": (pr128 or {}).get("mergedAt"), "mergedBy": ((pr128 or {}).get("mergedBy") or {}).get("login"), "headRefOid": (pr128 or {}).get("headRefOid"), "err": e1, "pass": pr128_ok, "raw_gh_snapshot": pr128, # 오프라인 감사용 원본 응답 박제 }, "check_4_pr102_preserved": { "state": (pr102 or {}).get("state"), "headRefOid": (pr102 or {}).get("headRefOid"), "expected_head": PR102_HEAD, "err": e2, "pass": pr102_ok, "raw_gh_snapshot": pr102, # 오프라인 감사용 원본 응답 박제 }, "gh_observed_at_utc": obs_end, "gh_observation_note": "raw_gh_snapshot = runner 실행 환경(네트워크 가용)" "에서 gh 실호출 응답 원본. 무네트워크 감사 환경에서도" " 본 박제로 독립 대조 가능.", "all_read_only": True, "reconcile_pass": reconcile_pass, } _write_json(EV / f"{TASK}.reconcile.json", obj) if not reconcile_pass: raise Hold("reconcile_failed", { "contains": contains, "diff_ok": diff_ok, "pr128_ok": pr128_ok, "pr102_ok": pr102_ok}) return obj # ─── 3. closeout evidence — 9-R.4 8-key 독립 재검증 ─────────────────────────── def _f2_region_sha(rev: str) -> str: """task-2553 계열 정준 F2 정의 = owner_trigger_pat.py lines 119-156 (task-2553+12.pre-merge-gate / run_pr128_merge_closeout.py 와 동일). `git show :...| sed -n '119,156p' | sha256sum` 와 byte 동치.""" src = _git("show", f"{rev}:anu_v2/owner_trigger_pat.py").stdout lines = src.split("\n") lo, hi = F2_LINES region = "\n".join(lines[lo - 1:hi]) + "\n" # 1-based inclusive, sed 동치 return _sha256_bytes(region.encode()) def closeout_evidence(reconcile_obj: dict) -> dict: keys = {} # k1 merge_commit (gh 재확인) pr128, _ = _gh_json("pr", "view", str(PR), "--json", "state,mergeCommit,mergedAt") keys["merge_commit"] = { "value": ((pr128 or {}).get("mergeCommit") or {}).get("oid"), "expect": MERGE_COMMIT, "proven": bool(pr128) and ( (pr128.get("mergeCommit") or {}).get("oid") == MERGE_COMMIT and pr128.get("state") == "MERGED"), } # k2 merged_at keys["merged_at"] = { "value": (pr128 or {}).get("mergedAt"), "expect": MERGED_AT, "proven": bool(pr128) and pr128.get("mergedAt") == MERGED_AT, } # k3 effective_diff_6 (실측 재대조) df = _git("diff", "--name-only", BASE_PARENT, MERGE_COMMIT) actual_6 = sorted(x for x in df.stdout.splitlines() if x.strip()) keys["effective_diff_6"] = { "value": actual_6, "expect": EXPECTED_6, "proven": actual_6 == EXPECTED_6, } # k4 ci_11_pass (gh checks 재조회) rollup, _ = _gh_json("pr", "view", str(PR), "--json", "statusCheckRollup") rl = (rollup or {}).get("statusCheckRollup") or [] states = [(c.get("conclusion") or c.get("state")) for c in rl] names = {(c.get("name") or c.get("context")) for c in rl} ci_ok = (len(rl) == CI_TOTAL_EXPECT and all(s == "SUCCESS" for s in states) and set(CI_REQUIRED_UNIQUE).issubset(names)) keys["ci_11_pass"] = { "total": len(rl), "expect_total": CI_TOTAL_EXPECT, "all_success": all(s == "SUCCESS" for s in states), "required_covered": set(CI_REQUIRED_UNIQUE).issubset(names), "proven": ci_ok, } # k5 gemini_resolved (reviewThreads unresolved==0 재조회) q = ('query{repository(owner:"Jeon-Jonghyuk",name:"dev_workspace")' '{pullRequest(number:%d){reviewThreads(first:100)' '{totalCount nodes{isResolved}}}}}' % PR) g, _ = _gh_json("api", "graphql", "-f", f"query={q}") try: nodes = (g or {})["data"]["repository"]["pullRequest"][ "reviewThreads"]["nodes"] unresolved = sum(1 for n in nodes if not n["isResolved"]) gem_ok = unresolved == 0 except Exception: # noqa: BLE001 unresolved, gem_ok = None, False keys["gemini_resolved"] = { "unresolved": unresolved, "proven": gem_ok, } # k6 F1 RED→GREEN (merge commit 의 evidence log 존재·내용) red = _git("show", f"{MERGE_COMMIT}:memory/events/task-2553+1.red-evidence.log") grn = _git("show", f"{MERGE_COMMIT}:memory/events/task-2553+1.green-evidence.log") red_t, grn_t = red.stdout, grn.stdout f1_ok = (red.returncode == 0 and grn.returncode == 0 and "RED" in red_t and "FAILED" in red_t and "GREEN" in grn_t and "PASSED" in grn_t and "F1" in (red_t + grn_t)) keys["f1_red_green"] = { "source": f"git show {MERGE_COMMIT[:12]}:memory/events/" f"task-2553+1.{{red,green}}-evidence.log (머지커밋판 — " f"live workspace 파일 아님; +11 내용 오독 방지)", "red_present": red.returncode == 0, "green_present": grn.returncode == 0, "red_has_FAILED": "FAILED" in red_t, "green_has_PASSED": "PASSED" in grn_t, "red_snippet": red_t[:240], "green_snippet": grn_t[:240], "proven": f1_ok, } # k7 F2 byte-identical (PR#102==merged F2-region sha, expected 와 일치) f2_pr102 = _f2_region_sha(PR102_HEAD) f2_merged = _f2_region_sha(MERGE_COMMIT) f2_ok = (f2_pr102 == f2_merged == F2_REGION_SHA_EXPECT) keys["f2_byte_identical"] = { "pr102_f2_region_sha": f2_pr102, "merged_f2_region_sha": f2_merged, "expect": F2_REGION_SHA_EXPECT, "proven": f2_ok, } # k8 preserved tests 무수정 (effective diff 에 신규 1파일만, 보존 0) preserved_touched = [p for p in actual_6 if any(h in p for h in PRESERVED_TEST_HINTS)] new_test_in_diff = NEW_TEST_FILE in actual_6 pres_ok = (len(preserved_touched) == 0 and new_test_in_diff and f2_ok) keys["preserved_tests_unmodified"] = { "preserved_test_files_in_diff": preserved_touched, "new_regression_test_in_diff": new_test_in_diff, "proven": pres_ok, } all_proven = all(v.get("proven") for v in keys.values()) obj = { "task": TASK, "ts": _utc(), "subject": "task-2553+1 (F1-solo) 본질 완료기준 ↔ PR#128 merge evidence", "verification_doctrine": "9-R.4 — 각 키가 closeout.json present AND " "개별 evidence 로 runner 독립 재증명. " "evidence_consistent boolean 단독 신뢰 금지.", "keys": keys, "reconcile_ref": "memory/events/%s.reconcile.json" % TASK, "reconcile_pass": reconcile_obj["reconcile_pass"], "all_keys_independently_proven": all_proven, "evidence_consistent": all_proven and reconcile_obj["reconcile_pass"], } return obj # ─── 4. schema-safe closeout — DRY 검증(무-write) / COMMIT(무-HOLD) 분리 ────── # CRITICAL-1 fail-closed: 모든 HOLD 가능 검증을 closeout/.done write 이전에 # 완료. write 단계는 HOLD 정적 부재(이미 검증된 데이터만 디스크화). NEW_KEY = "closeout_2553p13" MD_MARKER = "## task-2553+13 closeout" def _validate_result_json_append(block: dict): """검증만(write 0): parse + **사전존재 키(NEW_KEY 제외) 1개도 변형·삭제 0** 확인. 실패 → HOLD. 반환: (payload_str, mode). 9-R.3 "기존 key 변형 0" 의 '기존 key' = task-2553+1 사전존재 23 키. `NEW_KEY`(closeout_2553p13) 는 본 task 가 추가/소유하는 namespace 로, 동일 task 내 교정 재실행 시 **자기 키 갱신**은 허용(사전 23키는 불변). block 은 결정적(merge evidence 기반, wall-clock 미포함)이라 동일 evidence 재실행 시 byte-동일 → idempotent. """ p = EV / "task-2553+1.result.json" try: raw = p.read_text(encoding="utf-8") except FileNotFoundError: raise Hold("result_json_missing", str(p)) try: data = json.loads(raw) except json.JSONDecodeError as e: raise Hold("result_json_parse_failed", str(e)) if not isinstance(data, dict): raise Hold("result_json_not_object", type(data).__name__) # 사전존재 키(NEW_KEY 제외) canonical 스냅샷 pre = {k: json.dumps(v, sort_keys=True, ensure_ascii=False) for k, v in data.items() if k != NEW_KEY} had_key = NEW_KEY in data new_data = dict(data) new_data[NEW_KEY] = block post = {k: json.dumps(new_data[k], sort_keys=True, ensure_ascii=False) for k in new_data if k != NEW_KEY} if post != pre: # 사전 23키 변형·삭제 감지 → HOLD raise Hold("result_json_existing_key_mutation_detected", sorted(set(pre) ^ set(post))) payload = json.dumps(new_data, ensure_ascii=False, indent=2) + "\n" if had_key and json.dumps(data[NEW_KEY], sort_keys=True) == json.dumps( block, sort_keys=True): mode = {"mode": "idempotent_noop", "key": NEW_KEY, "preserved_keys": sorted(pre)} elif had_key: mode = {"mode": "own_key_corrective_update", "key": NEW_KEY, "preserved_keys": sorted(pre)} else: mode = {"mode": "top_level_key_added", "key": NEW_KEY, "preserved_keys": sorted(pre)} return payload, mode def _validate_md_append(): """검증만(write 0): marker 충돌 확인, append payload 산출.""" p = RP / "task-2553+1.md" try: orig = p.read_text(encoding="utf-8") except FileNotFoundError: raise Hold("md_missing", str(p)) if MD_MARKER in orig: # idempotent: 원본 그대로 반환(무조건 write 가 byte 무변경) return orig, {"mode": "idempotent_noop"} section = ( f"\n\n{MD_MARKER} — F1-solo evidence-based 완료 박제 ({_utc()})\n\n" f"PR #128 BOT merge SUCCESS(mergeCommit `{MERGE_COMMIT}`, mergedAt " f"`{MERGED_AT}`, 비가역)로 task-2553+1(F1-solo) 본질 완료기준이 merge " f"evidence 와 정합 확인됨. post-merge smoke 실패는 테스트 로더 " f"sys.modules 미등록 harness false-negative(머지코드 정상 — CI 11/11 " f"+ 격리 2-branch 결정적 재현). 8 완료기준 키(merge_commit / merged_at " f"/ effective_diff_6 / ci_11_pass / gemini_resolved / F1 RED→GREEN / " f"F2 byte-identical / preserved tests 무수정) 전건 runner 가 source " f"에서 독립 재계산 PASS. 단일 권위 = " f"`memory/events/task-2553+1.closeout.json` (9-R.3 sidecar). " f"result.json 은 최상위 key `{NEW_KEY}` 비파괴 추가. .done = " f"source 재계산 게이트 통과 후 생성(9-R.4). 본 섹션 EOF append " f"(기존 라인 무변경).\n") return (orig + section), {"mode": "eof_section_appended", "bytes_added": len(section)} def _reverify_keys_from_source(pin: str) -> dict: """9-R.4 / CRITICAL-2: stored boolean 신뢰 0 — 8 완료기준 키를 git/gh source 에서 **다시 독립 재계산**하고 EXPECTED 상수와 직접 대조. 1개라도 미증명 → Hold. """ proofs = {} pr128, _ = _gh_json("pr", "view", str(PR), "--json", "state,mergeCommit,mergedAt") mc = ((pr128 or {}).get("mergeCommit") or {}).get("oid") proofs["merge_commit"] = (bool(pr128) and (pr128 or {}).get("state") == "MERGED" and mc == MERGE_COMMIT) proofs["merged_at"] = bool(pr128) and ( pr128 or {}).get("mergedAt") == MERGED_AT df = _git("diff", "--name-only", BASE_PARENT, MERGE_COMMIT) actual_6 = sorted(x for x in df.stdout.splitlines() if x.strip()) proofs["effective_diff_6"] = actual_6 == EXPECTED_6 rollup, _ = _gh_json("pr", "view", str(PR), "--json", "statusCheckRollup") rl = (rollup or {}).get("statusCheckRollup") or [] states = [(c.get("conclusion") or c.get("state")) for c in rl] names = {(c.get("name") or c.get("context")) for c in rl} proofs["ci_11_pass"] = (len(rl) == CI_TOTAL_EXPECT and all(s == "SUCCESS" for s in states) and set(CI_REQUIRED_UNIQUE).issubset(names)) q = ('query{repository(owner:"Jeon-Jonghyuk",name:"dev_workspace")' '{pullRequest(number:%d){reviewThreads(first:100)' '{nodes{isResolved}}}}}' % PR) g, _ = _gh_json("api", "graphql", "-f", f"query={q}") try: nodes = (g or {})["data"]["repository"]["pullRequest"][ "reviewThreads"]["nodes"] proofs["gemini_resolved"] = sum( 1 for n in nodes if not n["isResolved"]) == 0 except Exception: # noqa: BLE001 proofs["gemini_resolved"] = False red = _git("show", f"{MERGE_COMMIT}:memory/events/task-2553+1.red-evidence.log") grn = _git("show", f"{MERGE_COMMIT}:memory/events/task-2553+1.green-evidence.log") proofs["f1_red_green"] = ( red.returncode == 0 and grn.returncode == 0 and "FAILED" in red.stdout and "PASSED" in grn.stdout and "F1" in (red.stdout + grn.stdout)) f2_a = _f2_region_sha(PR102_HEAD) f2_b = _f2_region_sha(MERGE_COMMIT) proofs["f2_byte_identical"] = (f2_a == f2_b == F2_REGION_SHA_EXPECT) preserved_touched = [p for p in actual_6 if any(h in p for h in PRESERVED_TEST_HINTS)] proofs["preserved_tests_unmodified"] = ( not preserved_touched and NEW_TEST_FILE in actual_6 and proofs["f2_byte_identical"]) # pinned reconcile ref 동치(단일 observation window 일관성) proofs["reconcile_pin_consistent"] = ( _git("rev-parse", "origin/main").stdout.strip() == pin or pin in _git("rev-list", "origin/main").stdout.split()) unproven = [k for k, v in proofs.items() if not v] if unproven: raise Hold("done_gate_source_reverify_failed", {"unproven": unproven, "proofs": proofs}) return proofs def build_closeout(ev: dict) -> dict: return { "task": "task-2553+1", "closeout_by": TASK, "ts": _utc(), "authority": "SINGLE — 본 sidecar 가 task-2553+1 closeout 단일 권위 " "(9-R.3). result.json 은 비파괴 key 추가만.", "subject": "task-2553+1 (F1-solo) — PR #128 merge 로 본질 완료", "merge_evidence": { "merge_commit": MERGE_COMMIT, "merged_at": MERGED_AT, "merged_by": MERGED_BY, "pr": PR, "pr_state": "MERGED", "irreversible": True, }, "completion_criteria_mapping": ev["keys"], "reconcile_pass": ev["reconcile_pass"], "all_keys_independently_proven": ev["all_keys_independently_proven"], "evidence_consistent": ev["evidence_consistent"], "classification_ref": f"memory/events/{TASK}.harness-artifact-classification.json", "reconcile_ref": f"memory/events/{TASK}.reconcile.json", "chair_option1_basis": "task-2553+1.chair-option1-f1solo-completion-" "packet.json (회장 Option 1, F1 solo GO)", } # ─── orchestrate ───────────────────────────────────────────────────────────── def main() -> int: started_epoch = time.time() started = _utc() ws_before = _ws_state() # activation-decision _write_json(EV / f"{TASK}.activation-decision.json", { "task": TASK, "ts": started, "track": "A", "executor": "dev5-team 마르둑 (key 109fa85250c6d46b) 1회 한정", "decision": "PROCEED — 회장 병행 GO, 선행 task-2553+12 merge SUCCESS " "(d08b8b0e 확정). Track B 완료 비대기.", "namespace": "task-2553+13.* / task-2553+1.closeout* 한정 (9-R.5)", "live_workspace_before": ws_before, }) try: # 9-R.5 namespace isolation = 본 runner 는 task-2553+13.* / # task-2553+1.closeout* 만 read/write. 병행 Track B 산출 경로를 # 정적으로 미참조(glob·open·인용 0) → cross-task contamination 차단. if ws_before["branch"] != LIVE_WS_EXPECT_BRANCH or \ ws_before["head"] != LIVE_WS_EXPECT_HEAD: raise Hold("live_workspace_drift_pre", ws_before) cls = classify() rec = reconcile() ev = closeout_evidence(rec) _write_json(EV / f"{TASK}.closeout-evidence.json", ev) if not ev["evidence_consistent"]: raise Hold("closeout_evidence_inconsistent", ev["keys"]) # ── CRITICAL-1: 모든 HOLD 가능 검증을 closeout/.done write 이전 완료 ── # (a) closeout sidecar payload 준비(write 0) closeout_obj = build_closeout(ev) closeout_payload = json.dumps(closeout_obj, ensure_ascii=False, indent=2) + "\n" # (b) result.json 비파괴 append DRY 검증(parse·mutation·conflict → HOLD) result_payload, res_mode = _validate_result_json_append({ "closeout_ref": "memory/events/task-2553+1.closeout.json", "closeout_by": TASK, "merge_commit": MERGE_COMMIT, "merged_at": MERGED_AT, "evidence_consistent": ev["evidence_consistent"], "note": "단일 권위 = sidecar closeout.json. 본 key 는 결정적 " "포인터(wall-clock 미포함 — 동일 evidence 재실행 시 " "byte-동일/idempotent). 타임스탬프는 sidecar 권위.", }) # (c) md EOF append DRY 검증 md_payload, md_mode = _validate_md_append() # (d) 9-R.4 / CRITICAL-2: .done 게이트 = source 재계산(stored boolean # 신뢰 0). 미증명 1개라도 → HOLD (closeout/.done write 0). done_proofs = _reverify_keys_from_source(rec["observation_window"][ "pinned_ref_sha"]) # (d2) closeout payload 의 8키 present 를 in-memory 사전 확인(HOLD-able) req8 = ["merge_commit", "merged_at", "effective_diff_6", "ci_11_pass", "gemini_resolved", "f1_red_green", "f2_byte_identical", "preserved_tests_unmodified"] ccm = closeout_obj.get("completion_criteria_mapping", {}) if not all(k in ccm and ccm[k].get("proven") for k in req8): raise Hold("closeout_payload_keys_incomplete", {"missing_or_unproven": [ k for k in req8 if not ( k in ccm and ccm[k].get("proven"))]}) if not all(done_proofs.values()): raise Hold("done_gate_source_reverify_failed", done_proofs) # (e) 최종 live workspace invariant 재확인(write 직전, drift → HOLD) ws_mid = _ws_state() if ws_mid != ws_before: raise Hold("live_workspace_drift_pre_commit", {"before": ws_before, "mid": ws_mid}) # ── COMMIT: 모든 HOLD 가능 검증 통과 완료. 이하 조건분기·raise Hold # 정적 부재 — 항상 4개 산출을 무조건 순서대로 디스크화. validator 가 # idempotent 시 원본 byte 그대로 반환하므로 무조건 write 가 안전. ── done_obj = { "task_id": "task-2553+1", "team_id": TEAM, "end_time": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(time.time() - started_epoch, 6), "qc_result": "PASS", "closeout_by": TASK, "evidence_gated": True, "gate": "9-R.4 — source(git/gh) 8키 독립 재계산 PASS + closeout " "payload 8키 present/proven (stored boolean 단독 신뢰 아님)", "source_reverify_proofs": done_proofs, "closeout_authority": "memory/events/task-2553+1.closeout.json", } done_payload = json.dumps(done_obj, ensure_ascii=False, indent=2) + "\n" _atomic_write(EV / "task-2553+1.closeout.json", closeout_payload) _atomic_write(EV / "task-2553+1.result.json", result_payload) _atomic_write(RP / "task-2553+1.md", md_payload) _atomic_write(EV / "task-2553+1.done", done_payload) dn = {"done": "memory/events/task-2553+1.done", "lifecycle": ".done → (외부 acker) .done.acked 준수", "source_reverified": True} cw = {"closeout_json": "memory/events/task-2553+1.closeout.json", "result_json": res_mode, "md": md_mode} # postflight: 정보 기록만(비-HOLD). 본 runner 는 git mutation 정적 # 부재(_git banned-list)이므로 marker write 가 HEAD/branch 변경 불가 # — postflight 는 그 불변을 확인·기록(드리프트 시 result 에 플래그). ws_after = _ws_state() ws_invariant_ok = (ws_after == ws_before and ws_after["branch"] == LIVE_WS_EXPECT_BRANCH and ws_after["head"] == LIVE_WS_EXPECT_HEAD) result = { "task": TASK, "ts": _utc(), "started": started, "status": "DONE", "track_a_standalone": "COMPLETE — 병행 Track B 완료 비대기 명시 " "(본 트랙 산출은 Track B artifact 인용 0)", "classification": cls["classification"], "classification_path": f"memory/events/{TASK}.harness-artifact-classification.json", "reconcile_path": f"memory/events/{TASK}.reconcile.json", "reconcile_pass": rec["reconcile_pass"], "closeout": cw, "done": dn, "pr128": {"state": "MERGED", "mergeCommit": MERGE_COMMIT, "mergedBy": MERGED_BY, "mergedAt": MERGED_AT}, "pr102_preserved": {"state": "OPEN", "head": PR102_HEAD}, "effective_diff_6": EXPECTED_6, "evidence_consistent": ev["evidence_consistent"], "all_keys_independently_proven": ev["all_keys_independently_proven"], "live_workspace_invariant": { "before": ws_before, "after": ws_after, "assertEqual_pass": ws_invariant_ok, "expected_branch": LIVE_WS_EXPECT_BRANCH, "expected_head": LIVE_WS_EXPECT_HEAD, "note": "git mutation 정적 부재 — marker write 는 HEAD/branch " "불변. postflight 확인·기록(비-HOLD).", }, "namespace_isolation_9R5": "task-2553+13.* / task-2553+1.closeout* " "한정. Track B artifact 인용 0.", "forbidden_not_performed": [ "rollback/revert/force/rebase 0", "admin override 0", "PR#128 재수정 0", "PR#102 원본 변경 0", "F2/phase3/mqe/_load_otp 변경 0", "production·test code 변경 0", "evidence 없는 closeout 0", "manual .done echo 0", "live workspace cleanup·reset·stash·rm 0", ], "hold": False, } _write_json(EV / f"{TASK}.result.json", result) return 0 except Hold as h: _emit_hold(h.reason, h.detail, started) _write_json(EV / f"{TASK}.result.json", { "task": TASK, "ts": _utc(), "started": started, "status": "HOLD_FOR_CHAIR", "hold": True, "reason": h.reason, "detail": h.detail, "closeout_created": False, "done_created": False, "live_workspace_invariant": { "before": ws_before, "after": _ws_state()}, }) return 2 if __name__ == "__main__": sys.exit(main())