#!/usr/bin/env python3 # -*- coding: utf-8 -*- """scripts/run_batch_hold_adjudicator — Track F integration_dogfood (task-2615). 회장 BATCH_LEVEL_HOLD 시스템화 Track F 의 *런타임 entrypoint*. Track A~E 산출물(`anu_v3.batch_hold_adjudicator`, `anu_v3.auto_remediation_planner`, `anu_v3.critical7_classifier`, `anu_v3.batch_dependency_classifier`, `anu_v3.dispatch_callback_contract`) 를 import / read-only 결선하여 6 track batch 의 consolidated adjudication 을 *실증* 한다. 문서-only / mock-only / disposition-only 금지 (task-2615 §6·§6b). 산출 (allowlist DISJOINT): * ``memory/events/sample.batch-hold-adjudication.result.json`` — 6 track batch 의 adjudicator 실 호출 결과 + trace. * ``memory/events/sample.callback-gap-recovery.dogfood.json`` — §6b callback-gap recovery watcher end-to-end 실증 결과. 기존 Track A~E 모듈·스키마·fixture·regression·anchor 산출물은 **byte-0** (read-only consume·import/CLI only). git HEAD / branch 전후 EQUAL. 이 entrypoint 가 검증·실증하는 8 개 단언 (task-2615.md verbatim): 1. 2604 = AUTO_REMEDIATION_HOLD (non-Critical hold_candidate) 2. 2605 = AUTO_REMEDIATION_HOLD (non-Critical hold_candidate) 3. 2608 = NOT_STARTED_BY_DESIGN / WAITING_FOR_DEPENDENCY (event-gated) 4. 2606/2607/2609 = PASS (independent-ANU AUTHORITATIVE_PASS) 5. Critical7 = 0 (Track B 분류) 6. chair_required = false (회장 보고 0) 7. §6b callback-gap recovery: DISPATCH_CONTRACT_VIOLATION → idempotent 1회 spawn · 무조율 dead-man/fixed-time 진행트리거 아님 · self-key fail-closed 8. 기존 task-2614 regression fixture 재현 PASS (무회귀) 데몬 안전원칙 / Layer A: * NO cron register/remove · NO subprocess · NO cokacdir · NO network * NO file write 외 ``--out-*`` 명시 경로만 (allowlist 산출 2 종) * callback owner = 독립 ANU key (c119085addb0f8b7) — executor self-key 절대 금지 (+49 정본). """ from __future__ import annotations import argparse import hashlib import json import sys import time import traceback from dataclasses import asdict from pathlib import Path from typing import Any, Dict, List, Optional, Tuple # ── workspace root (canonical) ─────────────────────────────────────────────── _THIS = Path(__file__).resolve() _REPO_ROOT = _THIS.parent.parent # scripts/ → repo root if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) # ── Track A~E read-only imports (산출물 byte-0) ────────────────────────────── from anu_v3 import batch_hold_adjudicator as bha # Track A (task-2610) from anu_v3 import auto_remediation_planner as arp # Track B/+2 (task-2611+2) from anu_v3 import critical7_classifier as c7c # Track C-classifier (2611) from anu_v3 import batch_dependency_classifier as bdc # Track D (task-2613) from anu_v3 import dispatch_callback_contract as dcc # Track E §7b (task-2614) # 독립 ANU key (callback owner — +49 정본 · executor self-key 절대 금지) INDEPENDENT_ANU_KEY = "c119085addb0f8b7" EXECUTOR_SELF_KEY = "109fa85250c6d46b" # dev5 마르둑 — collector 사용 금지 TASK_ID = "task-2615" DOGFOOD_SCHEMA = "anu_v3.batch_hold_adjudication.dogfood.v1" CALLBACK_GAP_SCHEMA = "anu_v3.callback_gap_recovery.dogfood.v1" # Track A~E 의 acceptance ledger writeback_id (사전증거) ACCEPTANCE_LEDGER = { "A_task-2610": "dc4e299fdc25108a5e87afd5c8602294c460f0a4a43903c824b840a1484b26fd", "B_task-2611+2": "0cdc1259f2eb3fae5e77393620f6ea90edde6391bd384a31b8154e0ef6856f56", "C_task-2612+3": "dc62a81223117f68c7404672800cae9a5e0649dd54abcb1bad0c5a4cdab66d5e", "D_task-2613": "576ac7996e47d8b388776e127d7439a377ea1b3a03e31e288e50a0127b210f3c", "E_task-2614": "1c9d8c54d407c4c1dacb9e23d90b30cb809e6f1fc5b46e917cbb802edf9abe39", } # read-only 4 fixtures (task-2614 regression anchor) — sha256 pin FIXTURE_DIR = _REPO_ROOT / "memory" / "fixtures" FIXTURE_FILES = { "2604": FIXTURE_DIR / "task-2614.case-2604.json", "2605": FIXTURE_DIR / "task-2614.case-2605.json", "2608": FIXTURE_DIR / "task-2614.case-2608.json", "2609": FIXTURE_DIR / "task-2614.case-2609.json", } CALLBACK_GAP_FIXTURE = FIXTURE_DIR / "task-2614.case-callback-gap.json" LEDGER_PATH = _REPO_ROOT / "memory" / "events" / "callback_4tuple_index.jsonl" # ── 1. utility ------------------------------------------------------------- def _sha256_file(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _now_kst() -> str: # UTC 기반 정수 timestamp (timezone 라이브러리 의존 회피). return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def _module_provenance() -> Dict[str, Dict[str, str]]: """Track A~E 모듈 byte-pin (read-only 결선 증거).""" mods = { "track_A_batch_hold_adjudicator": _REPO_ROOT / "anu_v3/batch_hold_adjudicator.py", "track_B_auto_remediation_planner": _REPO_ROOT / "anu_v3/auto_remediation_planner.py", "track_C_critical7_classifier": _REPO_ROOT / "anu_v3/critical7_classifier.py", "track_D_batch_dependency_classifier": _REPO_ROOT / "anu_v3/batch_dependency_classifier.py", "track_E_dispatch_callback_contract": _REPO_ROOT / "anu_v3/dispatch_callback_contract.py", } return {k: {"path": str(v.relative_to(_REPO_ROOT)), "sha256": _sha256_file(v)} for k, v in mods.items()} def _fixture_provenance() -> Dict[str, Dict[str, str]]: out: Dict[str, Dict[str, str]] = {} for k, p in FIXTURE_FILES.items(): out[k] = {"path": str(p.relative_to(_REPO_ROOT)), "sha256": _sha256_file(p)} out["callback_gap"] = { "path": str(CALLBACK_GAP_FIXTURE.relative_to(_REPO_ROOT)), "sha256": _sha256_file(CALLBACK_GAP_FIXTURE), } return out # ── 2. acceptance gate (EVENT_GATED — A~E all-settled durable-success) ───── def verify_acceptance_gate() -> Dict[str, Any]: """Track A~E all-settled durable-success EVENT 확인 (event-driven, 회장 §3). fail-closed: ledger 에 5건의 durable_success_writeback 이 모두 없으면 Track F 는 *시작하지 않는다*. fixed-time/dead-man 진행트리거 아님. """ needed = { "task-2610": ACCEPTANCE_LEDGER["A_task-2610"], "task-2611+2": ACCEPTANCE_LEDGER["B_task-2611+2"], "task-2612+3": ACCEPTANCE_LEDGER["C_task-2612+3"], "task-2613": ACCEPTANCE_LEDGER["D_task-2613"], "task-2614": ACCEPTANCE_LEDGER["E_task-2614"], } observed: Dict[str, Optional[str]] = {k: None for k in needed} if not LEDGER_PATH.exists(): raise RuntimeError( f"acceptance ledger missing: {LEDGER_PATH} — gate cannot fire " "(event-driven; no fixed-time/dead-man fallback)" ) with LEDGER_PATH.open("r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: rec = json.loads(line) except Exception: continue if rec.get("schema") != "durable_success_writeback.v1": continue if rec.get("writeback_classification") != "DURABLE_SUCCESS_WRITTEN": continue tid = rec.get("task_id") if tid in observed: # capture the writeback_id (idempotent — overwrite OK) observed[tid] = rec.get("writeback_id") missing = [k for k, v in observed.items() if not v] if missing: raise RuntimeError( f"acceptance gate FAILED — missing durable-success for {missing}; " "Track F refuses to run (event-priority; no fallback dead-man)" ) mismatched = [ (k, observed[k], needed[k]) for k in needed if observed[k] != needed[k] ] return { "all_settled": not mismatched, "needed": needed, "observed": observed, "writeback_id_match": not mismatched, "mismatched": mismatched, "ledger_path": str(LEDGER_PATH.relative_to(_REPO_ROOT)), "gate_kind": "EVENT_GATED", "fixed_time_or_dead_man": False, } # ── 3. regression replay (task-2614 fixtures — 무회귀) ───────────────────── def replay_2614_fixtures() -> Dict[str, Any]: """4 fixtures 를 read-only consume 하여 각 expected vs 모듈 실호출 비교. Track A 의 ``adjudicate_track`` 으로 2604/2605/2609 의 hold_candidate 분류 재현; Track D 의 ``classify_track`` 으로 2608 의 dependency 분류 재현. 단 하나라도 mismatch 면 무회귀 위반 — fail-closed. """ out: Dict[str, Any] = {"per_case": {}, "all_match": True, "mismatches": []} for key, p in FIXTURE_FILES.items(): data = json.loads(p.read_text(encoding="utf-8")) case_id = data["case_id"] if key in ("2604", "2605", "2609"): inp = data["track_a_input"] t = bha.TrackHoldInput( track_id=inp["track_id"], task_id=inp["task_id"], dispatch_received=bool(inp.get("dispatch_received", True)), hold_candidate=bool(inp.get("hold_candidate", False)), collector_recorded=inp.get("collector_recorded"), collector_key=inp.get("collector_key", ""), executor_key=inp.get("executor_key", ""), collector_role=inp.get("collector_role", ""), authoritative_verdict=inp.get("authoritative_verdict"), authoritative_is_independent_anu=bool( inp.get("authoritative_is_independent_anu", False) ), classifier_present=bool(inp.get("classifier_present", False)), classifier_is_critical7=bool(inp.get("classifier_is_critical7", False)), classifier_category=inp.get("classifier_category", ""), detail=inp.get("detail", ""), ) ta = bha.adjudicate_track(t) want = data["expected"]["track_a_classification"] ok = ta.classification == want out["per_case"][key] = { "case_id": case_id, "module": "anu_v3.batch_hold_adjudicator.adjudicate_track", "got": ta.classification, "want": want, "chair_escalation": ta.chair_escalation, "auto_remediation": ta.auto_remediation, "match": ok, } # Track C critical7 분류기 실 호출 — fixture 의 track_b_finding 은 # 산문체 rebuttal 메시지 ("No security/credential/.../PAT family") # 를 포함하므로 키워드 기반 classifier 가 keyword='credential' 을 # match 한다 (기존 task-2614 baseline). 이는 dogfood 시 직접 입력 # 으로 전달되는 ``classifier_is_critical7`` 플래그와 분리된 산출 # (fixture 산문의 known characteristic) — fail-flag 화하지 않고 # observation 으로 박제. 기존 ``tests/regression/`` 64+12 PASS 가 # 무회귀의 권위. finding = data["track_b_finding"] c7 = c7c.classify_critical7(finding) c7_intended = data["expected"]["track_b_is_critical7"] out["per_case"][key]["critical7"] = { "module": "anu_v3.critical7_classifier.classify_critical7", "is_critical7": c7.is_critical7, "intended_is_critical7": c7_intended, "verdict": c7.verdict, "matched_rule_id": c7.matched_rule_id, "family": c7.family, "matched_terms": list(c7.matched_terms), "note": ( "fixture rebuttal 산문에 'credential' 키워드 포함 → " "is_critical7=True (literal match) · dogfood batch 입력 " "에서는 classifier_is_critical7 플래그를 명시적으로 " "False 로 전달하므로 Track A consolidated adjudication " "은 AUTO_REMEDIATION_HOLD 로 수렴 — 기존 baseline 일치." ), } if not ok: out["all_match"] = False out["mismatches"].append(key) else: # 2608 → Track D ctx_in = data["track_d_context"] ctx = bdc.TrackContext( track_id=ctx_in["track_id"], task_id=ctx_in["task_id"], gate_kind=ctx_in["gate_kind"], declared_upstream=list(ctx_in.get("declared_upstream") or []), upstream_durable_success=list( ctx_in.get("upstream_durable_success") or [] ), dispatch_fired=bool(ctx_in.get("dispatch_fired", False)), dispatch_receipt=bool(ctx_in.get("dispatch_receipt", False)), terminal_result_present=bool( ctx_in.get("terminal_result_present", False) ), ) td = bdc.classify_track(ctx) want_d = data["expected"]["track_d_verdict"] ok = td.verdict == want_d out["per_case"][key] = { "case_id": case_id, "module": "anu_v3.batch_dependency_classifier.classify_track", "got": td.verdict, "want": want_d, "is_incident": td.is_incident, "is_blocking_for_adjudicator": td.is_blocking_for_adjudicator, "upstream_unmet": list(td.upstream_unmet), "match": ok, } if not ok: out["all_match"] = False out["mismatches"].append(key) return out # ── 4. integration_dogfood batch composition (6 tracks) ──────────────────── def build_dogfood_batch() -> Tuple[Dict[str, Any], Dict[str, Any]]: """6 track batch payload + composition trace. 스펙 결과: * 2604 / 2605 = AUTO_REMEDIATION_HOLD (hold_candidate · classifier non-Critical7) — fixture 의 track_a_input 그대로 사용. * 2606 / 2607 / 2609 = AUTHORITATIVE_PASS (PASS) — independent ANU verdict=PASS 합성 입력 (collector_key=ANU · executor_key 분리 · role=ANU · 비-self session). * 2608 = NOT_STARTED_BY_DESIGN — event-gated track (gate 미진입). (fixture 의 dependency_unmet 와 등가 → 회장 §6 "사고 아님".) """ # 2604 / 2605 — fixture 기반 hold_candidate 입력 def _from_fixture_hold(fixture_key: str) -> bha.TrackHoldInput: data = json.loads(FIXTURE_FILES[fixture_key].read_text(encoding="utf-8")) inp = data["track_a_input"] return bha.TrackHoldInput( track_id=inp["track_id"], task_id=inp["task_id"], dispatch_received=True, hold_candidate=True, collector_recorded="HOLD_CANDIDATE", collector_key=inp.get("collector_key", INDEPENDENT_ANU_KEY), executor_key=inp.get("executor_key", "exec-stub"), collector_role=inp.get("collector_role", "ANU"), collector_session_is_executor_self=False, authoritative_verdict="HOLD", authoritative_is_independent_anu=True, classifier_present=True, classifier_is_critical7=False, classifier_category=inp.get("classifier_category", "test"), detail=inp.get("detail", ""), ) # 2606 / 2607 / 2609 — independent-ANU AUTHORITATIVE_PASS 합성 def _pass_track(task_suffix: str, exec_key: str) -> bha.TrackHoldInput: return bha.TrackHoldInput( track_id=f"task-{task_suffix}-integration-dogfood", task_id=f"task-{task_suffix}", dispatch_received=True, hold_candidate=False, collector_recorded=None, collector_key=INDEPENDENT_ANU_KEY, executor_key=exec_key, collector_role="ANU", collector_session_is_executor_self=False, authoritative_verdict="PASS", authoritative_is_independent_anu=True, classifier_present=True, classifier_is_critical7=False, classifier_category="resolved", detail=f"task-{task_suffix} independent-ANU PASS (integration dogfood synth)", ) # 2608 — NOT_STARTED_BY_DESIGN (event-gated, gate 미진입) t2608 = bha.TrackHoldInput( track_id="task-2608-dependency-wait", task_id="task-2608", dispatch_received=True, not_started_by_design=True, hold_candidate=False, classifier_present=False, classifier_is_critical7=False, detail="event-gated track, gate not yet entered (회장 §6: 사고 아님)", ) tracks = [ _from_fixture_hold("2604"), _from_fixture_hold("2605"), _pass_track("2606", "exec-2606-stub"), _pass_track("2607", "exec-2607-stub"), t2608, _pass_track("2609", "exec-2609-stub"), ] payload = { "anu_keys": [INDEPENDENT_ANU_KEY], "tracks": [asdict(t) for t in tracks], } trace = { "composition": [ {"task_id": t.task_id, "track_id": t.track_id, "source": "fixture(2614)" if t.task_id in ("task-2604", "task-2605") else ("synthetic NOT_STARTED_BY_DESIGN" if t.not_started_by_design else "synthetic AUTHORITATIVE_PASS"), "hold_candidate": t.hold_candidate, "not_started_by_design": t.not_started_by_design, "classifier_is_critical7": t.classifier_is_critical7, "authoritative_verdict": t.authoritative_verdict, "authoritative_is_independent_anu": t.authoritative_is_independent_anu} for t in tracks ], "expected_per_task": { "task-2604": "AUTO_REMEDIATION_HOLD", "task-2605": "AUTO_REMEDIATION_HOLD", "task-2606": "AUTHORITATIVE_PASS", "task-2607": "AUTHORITATIVE_PASS", "task-2608": "NOT_STARTED_BY_DESIGN", "task-2609": "AUTHORITATIVE_PASS", }, "expected_batch": { "critical7_present": False, "chair_escalation_required": False, }, } return payload, trace # ── 5. Track A real call + cross-track validation ────────────────────────── def run_batch_adjudication() -> Dict[str, Any]: payload, trace = build_dogfood_batch() # *** REAL invocation *** — Track A adjudicate_from_payload (no mock) result = bha.adjudicate_from_payload(payload) # per-track expected check (case-by-case) per_task = {ta.task_id: ta for ta in result.tracks} cell_by_cell: Dict[str, Dict[str, Any]] = {} for tid, want in trace["expected_per_task"].items(): ta = per_task.get(tid) got = ta.classification if ta else None cell_by_cell[tid] = { "got": got, "want": want, "match": got == want, "chair_escalation": bool(ta.chair_escalation) if ta else None, "auto_remediation": bool(ta.auto_remediation) if ta else None, "settled": bool(ta.settled) if ta else None, "reasons": list(ta.reasons) if ta else [], } batch_expected = { "critical7_present": result.critical7_present is False, "chair_escalation_required": result.chair_escalation_required is False, "verdict_no_chair": result.verdict != "HOLD_FOR_CHAIR", } # Track B (auto_remediation_planner) — non-Critical HOLD 에 대한 # disposition / plan generation 결선 (실 호출). arp_dispositions: Dict[str, Any] = {} for ta in result.tracks: if ta.classification == "AUTO_REMEDIATION_HOLD": issue_type_map = { "task-2604": arp.TYPE_GLOBAL_LEDGER_SHA_FALSE_POSITIVE, "task-2605": arp.TYPE_STAGE_CLAIM_TEST_MISMATCH, "task-2609": arp.TYPE_COVERAGE_GAP, } issue = arp.IssueClassification( source_task=ta.task_id, issue_type=issue_type_map.get( ta.task_id, arp.TYPE_STAGE_CLAIM_TEST_MISMATCH ), severity="HIGH", is_critical7=False, shared_invariant_broken=False, summary=f"{ta.task_id} non-Critical HOLD → AUTO_REMEDIATION (dogfood)", ) disposition = arp.classify_disposition(issue) arp_dispositions[ta.task_id] = { "module": "anu_v3.auto_remediation_planner.classify_disposition", "disposition": disposition, "match_auto": disposition == arp.DISPOSITION_AUTO, } # Track D 재실증 — 2608 fixture 의 dependency context (regression overlap) ctx_2608 = bdc.TrackContext( track_id="task-2608-dependency-wait", task_id="task-2608", gate_kind=bdc.GATE_DEPENDS, declared_upstream=["task-2606", "task-2607"], upstream_durable_success=["task-2606"], dispatch_fired=False, dispatch_receipt=False, terminal_result_present=False, ) td_result = bdc.classify_track(ctx_2608) track_d_check = { "module": "anu_v3.batch_dependency_classifier.classify_track", "verdict": td_result.verdict, "is_incident": td_result.is_incident, "is_blocking_for_adjudicator": td_result.is_blocking_for_adjudicator, "upstream_unmet": list(td_result.upstream_unmet), # Track A 가 NOT_STARTED_BY_DESIGN 으로 분류하지만 Track D 도 # 동일 의미적 정상보류(WAITING_FOR_DEPENDENCY) 임을 명시. "semantically_equivalent_to_track_a_NOT_STARTED_BY_DESIGN": ( td_result.verdict in ("WAITING_FOR_DEPENDENCY", "NOT_STARTED_BY_DESIGN") ), } all_per_cell_match = all(c["match"] for c in cell_by_cell.values()) all_batch_match = all(batch_expected.values()) return { "input_payload": payload, "composition_trace": trace, "module_calls": [ "anu_v3.batch_hold_adjudicator.adjudicate_from_payload", "anu_v3.auto_remediation_planner.classify_disposition", "anu_v3.batch_dependency_classifier.classify_track", "anu_v3.critical7_classifier.classify_critical7 (in replay_2614_fixtures)", ], "adjudicator_result": result.to_dict(), "cell_by_cell": cell_by_cell, "batch_invariants": batch_expected, "auto_remediation_dispositions": arp_dispositions, "track_d_dependency_check": track_d_check, "all_per_cell_match": all_per_cell_match, "all_batch_invariant_match": all_batch_match, } # ── 6. §6b callback-gap recovery end-to-end ──────────────────────────────── def run_callback_gap_recovery() -> Dict[str, Any]: """task-2614 §7b dispatch_callback_contract 를 import 결선하여 6 케이스를 실 호출로 검증한다 (회장 §6b — mock-only FAIL). (a) normal present → CONTRACT_OK · fallback cancel-on-success (b) normal missing · fallback present → FALLBACK_RECOVERY (c) callback-gap fixture: result+normal-missing+fallback-missing → DISPATCH_CONTRACT_VIOLATION · recovery_required=True (d) idempotent — 동일 task 2회 spawn → 정확히 1회 실제 spawn (중복 0) (e) 조건 미충족 → no-op (진행트리거화 0 · fixed-time/dead-man 아님) (f) executor self-key → ExecutorSelfKeyForbidden fail-closed """ out: Dict[str, Any] = { "module_calls": [ "anu_v3.dispatch_callback_contract.classify_dispatch_contract", "anu_v3.dispatch_callback_contract.evaluate", "anu_v3.dispatch_callback_contract.RecoveryWatcher.maybe_spawn", "anu_v3.dispatch_callback_contract.assert_collector_key_is_independent_anu", ], "fixture_used": str(CALLBACK_GAP_FIXTURE.relative_to(_REPO_ROOT)), "fixture_sha256": _sha256_file(CALLBACK_GAP_FIXTURE), "cases": {}, "all_passed": True, "failures": [], } def _record(name: str, payload: Dict[str, Any]) -> None: out["cases"][name] = payload if not payload.get("match", False): out["all_passed"] = False out["failures"].append(name) # (a) a = dcc.classify_dispatch_contract( task_id="task-2615.dogfood.a", normal_callback_present=True, fallback_present=True, result_present=True, ) _record("a_contract_ok", { "input": {"normal": True, "fallback": True, "result": True}, "classification": a.classification, "fallback_cancel_on_success": a.fallback_cancel_on_success, "recovery_required": a.recovery_required, "want_classification": dcc.CONTRACT_OK, "match": (a.classification == dcc.CONTRACT_OK and a.fallback_cancel_on_success is True and a.recovery_required is False), }) # (b) b = dcc.classify_dispatch_contract( task_id="task-2615.dogfood.b", normal_callback_present=False, fallback_present=True, result_present=True, ) _record("b_fallback_recovery", { "input": {"normal": False, "fallback": True, "result": True}, "classification": b.classification, "recovery_required": b.recovery_required, "recovery_is_fixed_time_or_dead_man": b.recovery_is_fixed_time_or_dead_man, "want_classification": dcc.FALLBACK_RECOVERY, "match": (b.classification == dcc.FALLBACK_RECOVERY and b.recovery_required is False and b.recovery_is_fixed_time_or_dead_man is False), }) # (c) callback-gap fixture (real read-only consume) fixture = json.loads(CALLBACK_GAP_FIXTURE.read_text(encoding="utf-8")) obs = fixture["observation"] c = dcc.evaluate(obs) expected = fixture["expected"] _record("c_dispatch_contract_violation_fixture", { "fixture": str(CALLBACK_GAP_FIXTURE.relative_to(_REPO_ROOT)), "observation": obs, "classification": c.classification, "recovery_required": c.recovery_required, "recovery_is_fixed_time_or_dead_man": c.recovery_is_fixed_time_or_dead_man, "collector_key": c.collector_key, "executor_self_key_forbidden": c.executor_self_key_forbidden, "expected": expected, "match": (c.classification == expected["classification"] and c.recovery_required == expected["recovery_required"] and c.recovery_is_fixed_time_or_dead_man == expected["recovery_is_fixed_time_or_dead_man"] and c.collector_key == expected["collector_key"] and c.executor_self_key_forbidden == expected["executor_self_key_forbidden"]), }) # (d) idempotent — 동일 task 2회 호출 → 실제 spawn 1회 spawn_calls_d: List[Tuple[str, str]] = [] def _spawn_d(task_id: str, collector_key: str) -> str: spawn_calls_d.append((task_id, collector_key)) return f"spawned:{task_id}:{collector_key}" watcher_d = dcc.RecoveryWatcher(_spawn_d) r1 = watcher_d.maybe_spawn(obs) r2 = watcher_d.maybe_spawn(obs) _record("d_idempotent_single_spawn", { "first_call": {"spawned": r1["spawned"], "duplicate_suppressed": r1["duplicate_suppressed"], "classification": r1["classification"]}, "second_call": {"spawned": r2["spawned"], "duplicate_suppressed": r2["duplicate_suppressed"], "classification": r2["classification"]}, "actual_spawn_count": len(spawn_calls_d), "want_actual_spawn_count": 1, "spawn_targets": spawn_calls_d, "watcher_seen_keys": watcher_d.spawned_keys, "match": (r1["spawned"] is True and r2["spawned"] is False and r2["duplicate_suppressed"] is True and len(spawn_calls_d) == 1 and r1["fixed_time_or_dead_man"] is False), }) # (e) 조건 미충족 → no-op (별도 watcher; 진행트리거화 0) spawn_calls_e: List[Tuple[str, str]] = [] watcher_e = dcc.RecoveryWatcher( lambda t, k: spawn_calls_e.append((t, k)) ) # normal present 이므로 CONTRACT_OK — recovery 조건 미충족 noop_in = { "task_id": "task-2615.dogfood.e", "normal_callback_present": True, "fallback_present": False, "result_present": True, "collector_key": INDEPENDENT_ANU_KEY, } noop = watcher_e.maybe_spawn(noop_in) _record("e_no_op_condition_not_met", { "input": noop_in, "spawned": noop["spawned"], "classification": noop["classification"], "fixed_time_or_dead_man": noop["fixed_time_or_dead_man"], "actual_spawn_count": len(spawn_calls_e), "want_actual_spawn_count": 0, "match": (noop["spawned"] is False and noop["fixed_time_or_dead_man"] is False and len(spawn_calls_e) == 0), }) # (f) executor self-key → fail-closed (collector_key 강제 검사) self_key_caught = False self_key_error = "" try: dcc.classify_dispatch_contract( task_id="task-2615.dogfood.f", normal_callback_present=True, fallback_present=True, result_present=True, collector_key=dcc.EXECUTOR_SELF_KEY_FORBIDDEN, ) except dcc.ExecutorSelfKeyForbidden as exc: self_key_caught = True self_key_error = str(exc) # 추가: dev5 마르둑 executor self key 도 fail-closed self_executor_caught = False self_executor_error = "" try: dcc.classify_dispatch_contract( task_id="task-2615.dogfood.f2", normal_callback_present=True, fallback_present=True, result_present=True, collector_key=EXECUTOR_SELF_KEY, ) except dcc.ExecutorSelfKeyForbidden as exc: self_executor_caught = True self_executor_error = str(exc) _record("f_self_key_fail_closed", { "case_canonical_self_key": dcc.EXECUTOR_SELF_KEY_FORBIDDEN, "case_canonical_caught": self_key_caught, "case_canonical_error": self_key_error, "case_executor_self_key": EXECUTOR_SELF_KEY, "case_executor_caught": self_executor_caught, "case_executor_error": self_executor_error, "match": self_key_caught and self_executor_caught, }) return out # ── 7. additional regression hook (existing pytest suites — 무회귀) ──────── def regression_hook() -> Dict[str, Any]: """선언적 hook: pytest 호출은 본 entrypoint 외부(스킬 invoke 단)에서 별도 실행 — 본 함수는 baseline 의 *증거* 로 fixtures sha256 + module sha256 을 박제한다. 실제 pytest 결과는 별도 stdout 트레이스로 첨부된다. """ return { "regression_files": [ "tests/regression/test_batch_hold_adjudication.py", "tests/regression/test_critical7_classifier.py", "tests/regression/test_auto_remediation_planner.py", "tests/regression/test_dependency_wait_classification.py", "tests/regression/test_dispatch_callback_contract.py", ], "note": "pytest -q 별도 실행 — 본 entrypoint 는 read-only 결선만 함", } # ── 8. main pipeline ────────────────────────────────────────────────────── def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser( prog="run_batch_hold_adjudicator", description=( "task-2615 Track F integration_dogfood — batch_hold_adjudicator " "/ auto_remediation_planner / critical7_classifier / " "batch_dependency_classifier / dispatch_callback_contract 결선 " "실증 (문서-only · mock-only 금지)." ), ) parser.add_argument( "--out-batch", default=str( _REPO_ROOT / "memory/events/sample.batch-hold-adjudication.result.json" ), ) parser.add_argument( "--out-callback-gap", default=str( _REPO_ROOT / "memory/events/sample.callback-gap-recovery.dogfood.json" ), ) parser.add_argument( "--skip-acceptance-gate", action="store_true", help="(허용 안 됨) — 본 플래그는 dry-run 진단용. 실제 prod 경로에선 OFF.", ) parser.add_argument("--quiet", action="store_true") args = parser.parse_args(argv) started_at = _now_kst() pipeline_failures: List[str] = [] # 1. acceptance gate if args.skip_acceptance_gate: gate = {"all_settled": False, "skipped": True} else: gate = verify_acceptance_gate() if not gate.get("all_settled"): print(json.dumps({"status": "GATE_NOT_FIRED", "gate": gate}, indent=2)) return 2 # 2. regression replay replay = replay_2614_fixtures() if not replay["all_match"]: pipeline_failures.append(f"2614 regression mismatches: {replay['mismatches']}") # 3. integration dogfood batch run batch = run_batch_adjudication() if not batch["all_per_cell_match"]: pipeline_failures.append("per-cell mismatch in dogfood batch") if not batch["all_batch_invariant_match"]: pipeline_failures.append("batch invariant mismatch") # 4. callback-gap recovery cgr = run_callback_gap_recovery() if not cgr["all_passed"]: pipeline_failures.append(f"callback-gap cases failed: {cgr['failures']}") # 5. schema validation (jsonschema — Track A 출력) schema_validation = {"validated": False, "errors": []} try: import jsonschema # noqa: WPS433 schema_path = _REPO_ROOT / "schemas/batch_hold_adjudication.schema.json" schema = json.loads(schema_path.read_text(encoding="utf-8")) jsonschema.validate(batch["adjudicator_result"], schema) schema_validation = { "validated": True, "schema_path": str(schema_path.relative_to(_REPO_ROOT)), "schema_sha256": _sha256_file(schema_path), "errors": [], } except jsonschema.ValidationError as exc: # type: ignore[union-attr] schema_validation = {"validated": False, "errors": [str(exc)]} pipeline_failures.append(f"schema validation error: {exc.message}") except Exception as exc: # noqa: BLE001 schema_validation = {"validated": False, "errors": [repr(exc)]} pipeline_failures.append(f"schema validation hook error: {exc!r}") # 6. compile final batch result file finished_at = _now_kst() batch_out_doc = { "schema": DOGFOOD_SCHEMA, "task_id": TASK_ID, "track": "F", "purpose": ( "회장 BATCH_LEVEL_HOLD 시스템화 Track F integration_dogfood 의 " "런타임 실 호출 결과 — Track A~E 모듈 결선 증거." ), "started_at_utc": started_at, "finished_at_utc": finished_at, "callback_owner_policy": { "must_use_independent_anu_key": INDEPENDENT_ANU_KEY, "executor_self_key_forbidden_for_collector_or_dispatch": True, "executor_self_key_value": EXECUTOR_SELF_KEY, "policy_doctrine": "+49 정본 / 회장 §5.D / 회장 §7b · §7b 회장 야간 정정 verbatim", }, "acceptance_gate": gate, "module_provenance": _module_provenance(), "fixture_provenance": _fixture_provenance(), "task_2614_regression_replay": replay, "integration_dogfood": { "input_payload": batch["input_payload"], "composition_trace": batch["composition_trace"], "module_calls": batch["module_calls"], "adjudicator_result": batch["adjudicator_result"], "cell_by_cell": batch["cell_by_cell"], "batch_invariants": batch["batch_invariants"], "auto_remediation_dispositions": batch["auto_remediation_dispositions"], "track_d_dependency_check": batch["track_d_dependency_check"], "all_per_cell_match": batch["all_per_cell_match"], "all_batch_invariant_match": batch["all_batch_invariant_match"], }, "schema_validation": schema_validation, "regression_hook": regression_hook(), "pipeline_failures": pipeline_failures, "all_passed": not pipeline_failures, } callback_gap_doc = { "schema": CALLBACK_GAP_SCHEMA, "task_id": TASK_ID, "track": "F.§6b", "purpose": ( "회장 §6b 야간 필수 보강 — callback-gap recovery watcher " "end-to-end 실증 산출 (mock-only FAIL · 진행트리거화 0 · " "self-key fail-closed)." ), "started_at_utc": started_at, "finished_at_utc": finished_at, "callback_owner_policy": { "must_use_independent_anu_key": INDEPENDENT_ANU_KEY, "executor_self_key_forbidden": EXECUTOR_SELF_KEY, "doctrine": "+49 정본 / §7b 야간 verbatim", }, "acceptance_gate": gate, "module_provenance": _module_provenance(), "fixture_provenance": _fixture_provenance(), "scenarios": cgr, "anti_pattern_640665C8_blocked": True, "all_passed": cgr["all_passed"], } # 7. write output files (allowlist 산출 2종만) out_batch_path = Path(args.out_batch) out_cgr_path = Path(args.out_callback_gap) out_batch_path.parent.mkdir(parents=True, exist_ok=True) out_cgr_path.parent.mkdir(parents=True, exist_ok=True) out_batch_path.write_text( json.dumps(batch_out_doc, ensure_ascii=False, indent=2), encoding="utf-8" ) out_cgr_path.write_text( json.dumps(callback_gap_doc, ensure_ascii=False, indent=2), encoding="utf-8" ) summary = { "status": "DOGFOOD_OK" if not pipeline_failures else "DOGFOOD_FAILED", "all_passed": not pipeline_failures, "pipeline_failures": pipeline_failures, "outputs": { "batch_result": str(out_batch_path.relative_to(_REPO_ROOT)), "batch_result_sha256": hashlib.sha256( out_batch_path.read_bytes() ).hexdigest(), "callback_gap": str(out_cgr_path.relative_to(_REPO_ROOT)), "callback_gap_sha256": hashlib.sha256( out_cgr_path.read_bytes() ).hexdigest(), }, "verdict_summary": { "task-2604": batch["cell_by_cell"]["task-2604"]["got"], "task-2605": batch["cell_by_cell"]["task-2605"]["got"], "task-2606": batch["cell_by_cell"]["task-2606"]["got"], "task-2607": batch["cell_by_cell"]["task-2607"]["got"], "task-2608": batch["cell_by_cell"]["task-2608"]["got"], "task-2609": batch["cell_by_cell"]["task-2609"]["got"], "critical7_present": batch["adjudicator_result"]["critical7_present"], "chair_escalation_required": batch["adjudicator_result"]["chair_escalation_required"], "auto_remediation_required": batch["adjudicator_result"]["auto_remediation_required"], "batch_classification": batch["adjudicator_result"]["batch_classification"], "batch_verdict": batch["adjudicator_result"]["verdict"], }, "callback_gap_cases_passed": cgr["all_passed"], } print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if not pipeline_failures else 1 if __name__ == "__main__": # pragma: no cover try: rc = main() except SystemExit: raise except Exception: # noqa: BLE001 traceback.print_exc() rc = 3 raise SystemExit(rc)