# -*- coding: utf-8 -*- """task-2553+51 — TRACK 2 runtime structure SMOKE PILOT (read-only). 회장 3-track 배치 Track 2. goal_type = ``runtime_structure_smoke_pilot``. 단 하나의 짧은 read-only goal 로 +44 신구조(callback owner/key registry consistency)가 실제로 작동하는지 검증한다. 핵심 = 회장이 gate/HOLD/allowed/forbidden 을 풀어쓰지 않아도 ANU 가 default low-risk read-only profile 을 자동 산출하고 실 entrypoint 를 직접 호출(mock-only 금지)해 검증한다. 실 entrypoint (direct call, no mock): * ``anu_v3.callback_4tuple_registry.Callback4TupleRegistry`` — 영구 append-only 4-tuple ledger 를 READ-ONLY 로만 질의 (classify / history_for / validate_record / validate_identity). append / mark_completed 절대 호출 안 함 (ledger byte-0). * ``anu_v3.callback_owner_validator.validate_callback_owner_runtime`` — 독립 ANU key vs executor self key 로 owner/key/role fail-closed 검증. * ``anu_v3.authoritative_verdict_selector.select_authoritative_verdict`` — self-chain verdict 영구 비권위 + independent-ANU verdict 만 authoritative. Layer A / NO-CRON / READ-ONLY: ZERO cron register/remove, ZERO dispatch, ZERO merge, ZERO PR, ZERO branch/commit/push, ZERO credential, ZERO subprocess/cokacdir exec, ZERO ledger mutation, ZERO 기존 산출물 변조. 본 스크립트의 유일한 write 는 §6 allowlist 의 task-2553+51 신규 산출물뿐이며 그것도 ``--emit`` 시에만 수행한다. executor(이참나, key a999e2ea4c06d2fb) self-collector / self-adjudication / self-Codex / self-dispatch 0. authoritative verdict 는 독립 ANU collector(callback ANU key c119085addb0f8b7)가 산출하며 본 세션은 self-chain QUARANTINED 임을 *증명* 만 한다. """ from __future__ import annotations import argparse import importlib.util import json import sys from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional # scripts/ 는 canonical workspace root 바로 아래. ROOT = Path(__file__).resolve().parents[1] TASK_ID = "task-2553+51" GOAL_TYPE = "runtime_structure_smoke_pilot" # 회장 §10 verbatim — 독립 ANU collector key. executor self key 절대 금지. ANU_CALLBACK_KEY = "c119085addb0f8b7" EXECUTOR_SELF_KEY = "a999e2ea4c06d2fb" # dev7 이참나 (발사 금지 대상) CHAT_ID = "6937032012" # §6 allowlist (DISJOINT — Track1/3 와 0 overlap). 이 외 write 0. EXPECTED_FILES = ( "memory/events/task-2553+51.pilot-plan.json", "memory/events/task-2553+51.selected-policy-profile.json", "memory/events/task-2553+51.gate-hold-decision.json", "memory/events/task-2553+51.execution-result.json", "memory/events/task-2553+51.independent-collector-result.json", "memory/reports/task-2553+51.pilot-summary_260518.md", "scripts/runtime_smoke_pilot_2553plus51.py", "tests/regression/test_runtime_smoke_pilot_2553plus51.py", "memory/events/task-2553+51.decision.json", "memory/events/task-2553+51.result.json", ) def _load(modname: str, relpath: str): """Hermetic file-path import (collision-proof vs tests/ shadow).""" spec = importlib.util.spec_from_file_location(modname, ROOT / relpath) if spec is None or spec.loader is None: raise ImportError(f"cannot load {modname} from {relpath}") mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod # Pre-seed canonical dotted names so internal `from dispatch.* import` # resolves to the real workspace module (the validator imports the # enforcer by dotted name). _enf = _load("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") sys.modules["dispatch.callback_owner_enforcer"] = _enf _reg = _load("anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py") sys.modules["anu_v3.callback_4tuple_registry"] = _reg _val = _load("anu_v3.callback_owner_validator", "anu_v3/callback_owner_validator.py") _sel = _load( "anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py", ) Callback4TupleRegistry = _reg.Callback4TupleRegistry default_ledger_path = _reg.default_ledger_path NORMAL_CALLBACK_COMPLETED = _reg.NORMAL_CALLBACK_COMPLETED NO_LEDGER_RECORD = _reg.NO_LEDGER_RECORD validate_record = _reg.validate_record validate_callback_owner_runtime = _val.validate_callback_owner_runtime DEFAULT_ANU_KEYS = _enf.DEFAULT_ANU_KEYS is_anu_key = _enf.is_anu_key select_authoritative_verdict = _sel.select_authoritative_verdict VerdictRecord = _sel.VerdictRecord SELF_CHAIN_QUARANTINED = _sel.SELF_CHAIN_QUARANTINED # noqa: F841 (doc/ref) AUTHORITATIVE_VERDICT_PENDING = _sel.AUTHORITATIVE_VERDICT_PENDING AUTHORITATIVE_PASS = _sel.AUTHORITATIVE_PASS KIND_COLLECTOR_RESULT = _sel.KIND_COLLECTOR_RESULT @dataclass class Check: name: str entrypoint: str passed: bool expected: str observed: str detail: Dict[str, Any] = field(default_factory=dict) def to_json(self) -> Dict[str, Any]: return { "name": self.name, "entrypoint": self.entrypoint, "passed": self.passed, "expected": self.expected, "observed": self.observed, "detail": self.detail, } def check_a_registry_readonly() -> Check: """+44 registry 실 entrypoint READ-ONLY 질의 일관성. 실제 영구 ledger 를 query 메서드로만 읽는다 (append/mark_completed 호출 0 → ledger byte-0). 알려진 COMPLETED task 가 later-session 에서도 NORMAL_CALLBACK_COMPLETED 로 재확인되는지, 부재 task 가 fail-safe NO_LEDGER_RECORD 인지, 기록이 validate_record 를 통과하는지 검증. """ ledger = default_ledger_path(ROOT) reg = Callback4TupleRegistry(ledger) sampled: List[Dict[str, Any]] = [] completed_ok = True record_valid_ok = True for tid in ("task-2553+47", "task-2553+48"): hist = reg.history_for(tid) verdict = reg.classify(task_id=tid) latest = reg.latest_for(tid) rec_reasons = validate_record(latest) if latest is not None else ["NO_RECORD"] if not hist or verdict != NORMAL_CALLBACK_COMPLETED: completed_ok = False if rec_reasons: record_valid_ok = False sampled.append( { "task_id": tid, "history_lines": len(hist), "classify": verdict, "latest_status": getattr(latest, "status", None), "validate_record_reasons": rec_reasons, } ) absent_verdict = reg.classify(task_id="task-2553+51-NONEXISTENT-PROBE") failsafe_ok = absent_verdict == NO_LEDGER_RECORD passed = completed_ok and record_valid_ok and failsafe_ok return Check( name="A. callback_4tuple_registry read-only consistency", entrypoint="anu_v3.callback_4tuple_registry.Callback4TupleRegistry" ".{history_for,classify,latest_for,validate_record}", passed=passed, expected="known COMPLETED tasks -> NORMAL_CALLBACK_COMPLETED & " "validate_record clean; absent task -> NO_LEDGER_RECORD (fail-safe)", observed=( f"completed_ok={completed_ok} record_valid_ok={record_valid_ok} " f"failsafe_ok={failsafe_ok}" ), detail={ "ledger_path": str(ledger), "ledger_mutated": False, "sampled": sampled, "absent_probe_verdict": absent_verdict, }, ) def check_b_owner_validator() -> Check: """+49 owner/key/role fail-closed validator 실 entrypoint. 독립 ANU key(c119085addb0f8b7) collector -> PASS / registration allowed / owner_is_independent_anu. executor self key (a999e2ea4c06d2fb) collector -> FAIL / registration BLOCKED (self-collector 0 구조적 강제). """ common = dict( task_id=TASK_ID, executor_key=EXECUTOR_SELF_KEY, collector_role="ANU", normal_collector_cron_id="ANU-2553p51-NC", fallback_callback_cron_id="ANU-2553p51-FB", dispatch_cron_id="DG-2553p51-dev7-ichamna", chat_id=CHAT_ID, prompt_claims_anu_collector=True, entry_path="cokacdir_cron_direct", ) anu = validate_callback_owner_runtime( collector_key=ANU_CALLBACK_KEY, collector_owner_key=ANU_CALLBACK_KEY, **common, ) selfk = validate_callback_owner_runtime( collector_key=EXECUTOR_SELF_KEY, collector_owner_key=EXECUTOR_SELF_KEY, **common, ) anu_ok = ( anu.verdict == "PASS" and anu.registration_allowed and anu.owner_is_independent_anu ) self_blocked = ( selfk.verdict == "FAIL" and not selfk.registration_allowed and not selfk.owner_is_independent_anu ) passed = anu_ok and self_blocked return Check( name="B. callback_owner_validator fail-closed (ANU vs executor self)", entrypoint="anu_v3.callback_owner_validator." "validate_callback_owner_runtime", passed=passed, expected="ANU key collector -> PASS+allowed+independent; executor " "self key collector -> FAIL+blocked (self-collector structurally 0)", observed=f"anu_ok={anu_ok} self_blocked={self_blocked}", detail={ "anu_key_result": anu.to_json(), "executor_self_key_result": selfk.to_json(), }, ) def check_c_authoritative_selector() -> Check: """+49 authoritative verdict selector 실 entrypoint. executor self-session 산출물(self-chain) 은 PASS 라도 영구 비권위 (QUARANTINED / PENDING / verdict FAIL). 독립 ANU collector verdict (collector_key=ANU, role=ANU, !=executor, not self-session) 만 AUTHORITATIVE_PASS. """ anu_keys = sorted(DEFAULT_ANU_KEYS) self_chain = VerdictRecord( kind=KIND_COLLECTOR_RESULT, verdict="PASS", task_id=TASK_ID, executor_key=EXECUTOR_SELF_KEY, collector_key=EXECUTOR_SELF_KEY, collector_role="executor", session_is_executor_self=True, claimed_origin="independent_anu", # 거짓 주장 — 무권위 검증 detail="executor self-session pilot result (NEVER authoritative)", ) self_only = select_authoritative_verdict( [self_chain], task_id=TASK_ID, anu_keys=anu_keys ) independent = VerdictRecord( kind=KIND_COLLECTOR_RESULT, verdict="PASS", task_id=TASK_ID, executor_key=EXECUTOR_SELF_KEY, collector_key=ANU_CALLBACK_KEY, collector_role="ANU", session_is_executor_self=False, claimed_origin="independent_anu", detail="independent ANU collector verdict (authoritative)", ) with_independent = select_authoritative_verdict( [self_chain, independent], task_id=TASK_ID, anu_keys=anu_keys ) self_quarantined = ( self_only.classification == AUTHORITATIVE_VERDICT_PENDING and self_only.verdict == "FAIL" and self_only.authoritative_verdict is None and self_only.independent_anu_count == 0 ) independent_authoritative = ( with_independent.classification == AUTHORITATIVE_PASS and with_independent.verdict == "PASS" and with_independent.independent_anu_count == 1 ) passed = self_quarantined and independent_authoritative return Check( name="C. authoritative_verdict_selector (self-chain非권위)", entrypoint="anu_v3.authoritative_verdict_selector." "select_authoritative_verdict", passed=passed, expected="self-chain only -> PENDING/FAIL (no authoritative PASS); " "+independent ANU -> AUTHORITATIVE_PASS", observed=( f"self_quarantined={self_quarantined} " f"independent_authoritative={independent_authoritative}" ), detail={ "self_chain_only": self_only.to_json(), "with_independent_anu": with_independent.to_json(), }, ) def check_d_key_registry_consistency() -> Check: """callback owner/key registry 횡단 일관성 (회장 권고 pilot 후보). enforcer/validator 가 쓰는 독립 ANU key 집합이 회장 §10 가 강제하는 callback key 와 동일하고, executor self key 는 ANU key 가 아님을 is_anu_key 실 함수로 교차 확인. """ anu_keys = sorted(DEFAULT_ANU_KEYS) mandated_in_set = ANU_CALLBACK_KEY in DEFAULT_ANU_KEYS mandated_is_anu = is_anu_key(ANU_CALLBACK_KEY, anu_keys) self_not_anu = not is_anu_key(EXECUTOR_SELF_KEY, anu_keys) passed = mandated_in_set and mandated_is_anu and self_not_anu return Check( name="D. callback owner/key registry cross-module consistency", entrypoint="dispatch.callback_owner_enforcer.{DEFAULT_ANU_KEYS," "is_anu_key}", passed=passed, expected="mandated callback key == registered independent ANU key; " "executor self key NOT an ANU key", observed=( f"mandated_in_set={mandated_in_set} mandated_is_anu=" f"{mandated_is_anu} self_not_anu={self_not_anu}" ), detail={ "registered_anu_keys": anu_keys, "mandated_callback_key": ANU_CALLBACK_KEY, "executor_self_key": EXECUTOR_SELF_KEY, }, ) def run_pilot() -> Dict[str, Any]: """모든 실 entrypoint 를 직접 호출 (mock-only FAIL). read-only.""" checks = [ check_a_registry_readonly(), check_b_owner_validator(), check_c_authoritative_selector(), check_d_key_registry_consistency(), ] all_passed = all(c.passed for c in checks) # 이 verdict 는 executor self-chain 산출물 → 영구 비권위. # authoritative verdict 는 독립 ANU collector(callback)가 산출한다. return { "schema": "anu_v3.runtime_smoke_pilot.v1", "task_id": TASK_ID, "goal_type": GOAL_TYPE, "executor": "dev7-team 이참나", "executor_key": EXECUTOR_SELF_KEY, "mock_only": False, "real_entrypoints_called": True, "ledger_mutated": False, "self_chain_verdict": "PASS" if all_passed else "FAIL", "self_chain_authoritative": False, "authoritative_selection": "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR", "checks": [c.to_json() for c in checks], } def _write(rel: str, text: str) -> None: if rel not in EXPECTED_FILES: raise SystemExit(f"REFUSED: {rel} not in §6 allowlist (write 0 outside)") p = ROOT / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_text(text, encoding="utf-8") def main(argv: Optional[List[str]] = None) -> int: ap = argparse.ArgumentParser() ap.add_argument( "--emit", action="store_true", help="write §6 allowlisted artifacts (default: print only)", ) ns = ap.parse_args(argv) result = run_pilot() print(json.dumps(result, ensure_ascii=False, indent=2)) if ns.emit: _write( "memory/events/task-2553+51.execution-result.json", json.dumps(result, ensure_ascii=False, indent=2), ) return 0 if result["self_chain_verdict"] == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())