# -*- coding: utf-8 -*- """task-2605 — Track B: READ_ONLY_OPERATIONAL_PILOT (9-stage 자동 결선). 회장 §1 verbatim: 짧은 read-only goal 하나로 ANU 가 policy profile → dispatch planning → ANU-key callback → independent collector → registry → runtime loop → bounded enactor 흐름을 **실사용 검증**한다. 회장이 gate/HOLD/allowed/forbidden 을 풀어쓰지 않아도 ANU 가 policy_profile_engine + runtime guard 로 자동 산출·검증·통합한다. 핵심 = 본 스크립트가 각 단계의 **실 entrypoint 를 직접 호출**(mock-only FAIL · 문서-only 금지)해 9-stage 자동 진행을 입증한다. 입력은 task-2553 +57 의 짧은 read-only goal(``pilot_goal.json`` goal_request, goal_type = ``runtime_structure_smoke_pilot`` — "policy profile dry-run / callback owner·key registry consistency") 이며, +58 fallback acceptance criterion/non_blocking schema 를 read-only 로 소비한다. task-2605 고유 acceptance = **Track A(task-2604) durable-success EVENT 소비(event-driven, NOT fixed-time/dead-man)**: * A 미완 (ledger 에 task-2604 durable-success record 부재) → fallback criterion (b) REGISTRY_NON_BLOCKING_EXPLICIT_MARK 까지 허용 (OPERATIONAL_PASS). * A 완료 (ledger 에 task-2604 durable-success record 존재) → cancel-on-success live remove 확인 (criterion (a)) 또는 CANCEL_FAILED_CLASSIFIED (분류된 처분 — 미분류 잔존 아님). 실 entrypoint (direct call, no mock — stage ↔ module): 1. ``anu_v3.default_profile_resolver.run_default_profile_resolution`` / ``run_selected_profile_evidence`` — policy_profile_engine 정본 ``parse_goal_request -> resolve_policy`` READ-ONLY (engine byte-0). 2. ``anu_v3.callback_owner_validator.validate_callback_owner_runtime`` + ``assert_registration_permitted`` — ANU-key callback contract 강제 (executor self key → CallbackRegistrationBlocked). 3. executor read-only goal — RESOLVED 일관 + 미매핑 fail-closed + callback owner/key registry 일관(``dispatch.callback_owner_enforcer`` ``DEFAULT_ANU_KEYS``/``is_anu_key`` READ-ONLY). 4. ``anu_v3.self_collector_guard.guard_self_collector_session`` — normal callback owner 는 독립 ANU key 만; executor self key collector 는 ``SELF_COLLECTOR_FORBIDDEN``. 5. ``anu_v3.authoritative_verdict_selector.select_authoritative_verdict`` + ``anu_v3.cancel_on_success_live_wiring. lookup_fallback_from_durable_registry`` + ``scripts. validate_fallback_acceptance_2553plus58.evaluate_fallback_acceptance`` — independent-ANU 만 authoritative · self-chain QUARANTINED · Track A durable-success EVENT 소비로 acceptance criterion 자동 분기. 6. ``anu_v3.batch_settle_writeback.evaluate_durable_success_writeback`` — +53 durable-success write-back **평가만**(read-only). 실 append 는 독립 ANU collector 세션 소관(§7) — 본 executor 세션은 ledger 직접 write 0. 7. ``anu_v3.runtime_event_loop.RuntimeEventLoop.run`` — durable-registry 를 read-only 소비, Track A(task-2604) durable-success EVENT 즉시 감지. dead-man/fixed-time 진행 트리거 → 하드 FAIL 음성 입증. 8. ``anu_v3.runtime_event_enactor.RuntimeEventEnactor.enact`` — proposal -only · additive · merge/PR/write 0 (artifact_writer=None → dry). 9. consolidated result (+58 verdict · Track A event · subagent ledger). 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 기존 산출물 변조, ZERO ledger 직접 write. 본 스크립트의 유일한 durable write 는 §4 allowlist 의 task-2605 신규 산출물(``--emit`` 시에만) 뿐이다. ``callback_4tuple_index.jsonl`` 은 track 미소유 공유 durable 인프라 — 본 runner 는 read-only consume, direct write 0 (+53/+54 소관 mechanism / 독립 ANU collector 만 append). executor(dev6-team 페룬, key 1e41a2324a3ccdd0) self-collector / self-adjudication / self-Codex / self-dispatch / self-delegation 0. authoritative verdict·회수·검증·Codex·adjudication·batch 통합은 독립 ANU collector(callback ANU key c119085addb0f8b7)가 담당하며 본 self-chain 산출은 영구 비권위임을 *증명* 한다. """ from __future__ import annotations import argparse import importlib.util import json import sys import tempfile from pathlib import Path from typing import Any, Dict, List, Optional ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) TASK_ID = "task-2605" GOAL_TYPE = "runtime_structure_smoke_pilot" BATCH_ID = "task-2605-operational-pilot" TRACK_ID = "READ_ONLY_OPERATIONAL_PILOT" # 회장 §3/§5/§7 verbatim — 독립 ANU collector key. executor self key 절대 # 금지(+49 코드 정본 — 위반 시 CallbackRegistrationBlocked / # SELF_COLLECTOR_FORBIDDEN). ANU_CALLBACK_KEY = "c119085addb0f8b7" EXECUTOR_SELF_KEY = "1e41a2324a3ccdd0" # dev6-team 페룬 (발사/회수 금지 대상) CHAT_ID = "6937032012" TS_KST = "2026-05-19 04:30 KST" # 본 배치 ANU fallback — 안전망 한정 (진행/수락 트리거 아님). ANU_FALLBACK_AT_KST = "2026-05-19 07:45:00" FALLBACK_CRON_ID = f"ANU-fallback-safety-net:{TASK_ID}:{ANU_CALLBACK_KEY}" # Track A — task-2605 acceptance 가 event-driven 으로 소비하는 의존. TRACK_A_TASK_ID = "task-2604" # 선행 read-only consume (byte-0). +57 짧은 goal · +58 fallback 기준 · # task-2553+60 설계 산출물 전체. PLUS57_PILOT_GOAL = "memory/events/pilot_goal.json" PLUS57_EXEC_PLAN = "memory/events/pilot_execution_plan.json" PLUS57_SUCCESS = "memory/events/success_criteria.json" PLUS58_FALLBACK_CRIT = "memory/events/fallback_acceptance_criteria.json" PLUS58_NB_SCHEMA = "schemas/non_blocking_fallback_schema.json" PLUS60_PILOT = "scripts/run_operational_pilot_2553plus60.py" SHARED_LEDGER = "memory/events/callback_4tuple_index.jsonl" # §4 expected_files allowlist — 정확히 회장 명세 10종 (이 외 write 0, # additive only). callback_4tuple_index.jsonl 은 track 미소유 공유 # 인프라 — allowlist 아님(read-only consume). EXPECTED_FILES = ( "scripts/run_operational_pilot_2605.py", "memory/events/task-2605.pilot-run.json", "memory/events/task-2605.selected-profile.json", "memory/events/task-2605.execution-result.json", "memory/events/task-2605.independent-collector-result.json", "memory/events/task-2605.consolidated-result.json", "tests/regression/test_operational_pilot_2605.py", "memory/events/task-2605.decision.json", "memory/events/task-2605.result.json", "memory/reports/task-2605.md", ) # 본 runner 가 --emit 시 직접 write 하는 산출물. 스크립트/테스트/리포트는 # 정적(byte-0 자기보존). ledger 는 §4 별도 — track 직접 write 0. EMIT_FILES = ( "memory/events/task-2605.pilot-run.json", "memory/events/task-2605.selected-profile.json", "memory/events/task-2605.execution-result.json", "memory/events/task-2605.independent-collector-result.json", "memory/events/task-2605.consolidated-result.json", "memory/events/task-2605.decision.json", "memory/events/task-2605.result.json", ) class MockOnlyError(RuntimeError): """실 entrypoint 미사용(mock-only) 시 즉시 FAIL — 문서-only 금지.""" def _load_real(modname: str, relpath: str): """정본 모듈을 파일 경로로 로딩 (mock 0). repo 에 dispatch.py 와 dispatch/ 가 공존해 일반 import 가 컨텍스트에 따라 엇갈리므로 +60 pilot 과 동일하게 명시 경로 로딩.""" spec = importlib.util.spec_from_file_location(modname, ROOT / relpath) if spec is None or spec.loader is None: raise ModuleNotFoundError(f"real module not found: {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` / # `from anu_v3.* import` resolve to the real workspace module even under # pytest where ``dispatch.py`` shadows the ``dispatch/`` package (+60 # pilot 동형 — collision-proof, mock 0). _PRESEED = ( ("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py"), ( "dispatch.executor_completion_contract", "dispatch/executor_completion_contract.py", ), ("anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py"), ( "anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py", ), ) for _mn, _rp in _PRESEED: if _mn not in sys.modules: _load_real(_mn, _rp) def _read_json(rel: str) -> Any: return json.loads((ROOT / rel).read_text(encoding="utf-8")) def _consume_inputs() -> Dict[str, Any]: """+57 짧은 read-only goal + +58 fallback 기준 read-only consume. +60 설계는 정적 read-only 재사용(byte-0 — 본 runner 가 호출 0).""" pg = _read_json(PLUS57_PILOT_GOAL) goal_request = pg["pilot_goal"]["goal_request"] plus60_exists = (ROOT / PLUS60_PILOT).is_file() return { "pilot_goal_request": goal_request, "pilot_goal_title": pg["pilot_goal"]["title"], "pilot_goal_one_line": pg["pilot_goal"]["one_line_goal"], "exec_plan_schema": _read_json(PLUS57_EXEC_PLAN)["schema"], "success_criteria_ids": [ c["id"] for c in _read_json(PLUS57_SUCCESS)["pilot_success_criteria"] ], "fallback_criteria_schema": _read_json(PLUS58_FALLBACK_CRIT)["schema"], "nb_schema_id": _read_json(PLUS58_NB_SCHEMA)["$id"], "plus60_design_read_only_consumed": plus60_exists, } # ───────── Track A durable-success EVENT consumer (event-driven) ─────────── def _consume_track_a_durable_success_event() -> Dict[str, Any]: """task-2605 고유 — Track A(task-2604) durable-success EVENT 를 공유 durable ledger 에서 **read-only** 소비(event-driven, NOT fixed-time/ dead-man). ledger 직접 write 0. 정본 +44 registry entrypoint 로 소비.""" reg_mod = _load_real( "anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py", ) ledger = ROOT / SHARED_LEDGER durable_lines: List[Dict[str, Any]] = [] ledger_present = ledger.is_file() if ledger_present: for raw in ledger.read_text(encoding="utf-8").splitlines(): raw = raw.strip() if not raw: continue try: obj = json.loads(raw) except json.JSONDecodeError: continue if str(obj.get("task_id", "")) != TRACK_A_TASK_ID: continue if ( obj.get("schema") == "durable_success_writeback.v1" and obj.get("writeback_classification") == "DURABLE_SUCCESS_WRITTEN" ) or ( obj.get("schema") == "callback_4tuple_ledger_record.v1" and obj.get("role") == "anu_collector_durable_success" and obj.get("status") == "COMPLETED" ): durable_lines.append(obj) # +44 registry 정본으로도 교차 확인 (read-only). registry_record = None try: registry = reg_mod.Callback4TupleRegistry(ledger) rec = registry.latest_for(TRACK_A_TASK_ID) if rec is not None: registry_record = { "task_id": rec.task_id, "dispatch_cron_id": getattr(rec, "dispatch_cron_id", None), "chat_id": getattr(rec, "chat_id", None), "role": getattr(rec, "role", None), } except Exception as exc: # noqa: BLE001 - read-only consume 디커플 registry_record = {"lookup_error": str(exc)} a_complete = bool(durable_lines) or ( isinstance(registry_record, dict) and registry_record.get("task_id") == TRACK_A_TASK_ID and "lookup_error" not in registry_record and registry_record.get("role") == "anu_collector_durable_success" ) return { "real_entrypoint": "anu_v3.callback_4tuple_registry." "Callback4TupleRegistry.latest_for + shared ledger read-only scan " "(callback_4tuple_index.jsonl — track 미소유 공유 인프라, write 0)", "track_a_task_id": TRACK_A_TASK_ID, "shared_ledger_present": ledger_present, "track_a_durable_success_event_present": a_complete, "track_a_durable_records_count": len(durable_lines), "track_a_registry_record": registry_record, "event_driven": True, "fixed_time_or_dead_man_used": False, "track_a_status": "A_COMPLETE" if a_complete else "A_INCOMPLETE", } # ───────────────────────── 9-stage real-entrypoint chain ─────────────────── def _stage1_profile_selection(goal_request: Dict[str, Any]) -> Dict[str, Any]: """STAGE 1 — ANU policy profile resolve. policy_profile_engine 정본 read-only resolve. 회장 미지정 gate/HOLD/allowed/forbidden 자동 산출.""" res = _load_real( "anu_v3.default_profile_resolver", "anu_v3/default_profile_resolver.py" ) resolved = res.run_default_profile_resolution(goal_request) evidence = res.run_selected_profile_evidence(goal_request) # 음성: 미매핑 goal_type 은 RESOLVED 가 아니어야 함 (fail-closed). unknown = res.run_default_profile_resolution( {"goal_type": "__no_such_goal_type_2605__", "boundary": []} ) return { "real_entrypoint": "anu_v3.default_profile_resolver." "run_default_profile_resolution -> policy_profile_engine." "parse_goal_request -> resolve_policy (READ-ONLY, engine byte-0)", "resolved": resolved, "selected_profile_evidence": evidence, "unknown_goal_type_resolution": unknown, "auto_derived": True, "status": resolved.get("status"), "profile_bound": resolved.get("profile_bound"), } def _stage2_dispatch_planning(s1: Dict[str, Any]) -> Dict[str, Any]: """STAGE 2 — dispatch planning. ANU-key callback contract 강제. executor self key collector → CallbackRegistrationBlocked (구조적).""" cv = _load_real( "anu_v3.callback_owner_validator", "anu_v3/callback_owner_validator.py" ) common = dict( task_id=TASK_ID, executor_key=EXECUTOR_SELF_KEY, collector_role="ANU", normal_collector_cron_id=f"ANU-normal-callback:{ANU_CALLBACK_KEY}", fallback_callback_cron_id=FALLBACK_CRON_ID, dispatch_cron_id=f"dispatch:{TASK_ID}", chat_id=CHAT_ID, anu_keys=[ANU_CALLBACK_KEY], ) anu_ok = cv.validate_callback_owner_runtime( collector_key=ANU_CALLBACK_KEY, collector_owner_key=ANU_CALLBACK_KEY, **common, ) self_blocked = cv.validate_callback_owner_runtime( collector_key=EXECUTOR_SELF_KEY, collector_owner_key=EXECUTOR_SELF_KEY, **common, ) # hard gate — executor self key callback 등록은 구조적으로 차단. blocked_raised = False block_exc = "" try: cv.assert_registration_permitted(self_blocked) except cv.CallbackRegistrationBlocked as e: blocked_raised = True block_exc = str(e) return { "real_entrypoint": "anu_v3.callback_owner_validator." "validate_callback_owner_runtime + assert_registration_permitted", "anu_key_registration_allowed": anu_ok.registration_allowed, "anu_key_owner_is_independent_anu": anu_ok.owner_is_independent_anu, "self_key_registration_allowed": self_blocked.registration_allowed, "self_key_assert_raised_CallbackRegistrationBlocked": blocked_raised, "self_key_block_classification": self_blocked.classifications, "block_exception": block_exc, "callback_contract": { "normal_completion_callback_owner_key": ANU_CALLBACK_KEY, "executor_self_key_forbidden_as_owner": EXECUTOR_SELF_KEY, "next_action_basis": "Track A durable-success event " "(event-driven · NOT fixed-time/dead-man)", }, "profile_selected": s1["resolved"].get("resolved_profile_name"), } def _stage3_executor(s1: Dict[str, Any]) -> Dict[str, Any]: """STAGE 3 — executor read-only goal. RESOLVED 일관 + 미매핑 fail-closed + callback owner/key registry 일관.""" enf = _load_real( "dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py" ) anu_keys = sorted(enf.DEFAULT_ANU_KEYS) res = s1["resolved"] unk = s1["unknown_goal_type_resolution"] consistency = { "registered_anu_keys": anu_keys, "mandated_callback_key": ANU_CALLBACK_KEY, "executor_self_key": EXECUTOR_SELF_KEY, "mandated_is_anu_key": enf.is_anu_key(ANU_CALLBACK_KEY, anu_keys), "executor_self_is_anu_key": enf.is_anu_key(EXECUTOR_SELF_KEY, anu_keys), } goal_outcomes = { "resolved_consistent": res.get("status") == "RESOLVED" and res.get("profile_bound") is True, "unknown_fail_closed": unk.get("status") != "RESOLVED", "unknown_status": unk.get("status"), "callback_owner_key_consistent": ( consistency["mandated_is_anu_key"] is True and consistency["executor_self_is_anu_key"] is False ), } return { "real_entrypoint": "dispatch.callback_owner_enforcer." "{DEFAULT_ANU_KEYS,is_anu_key} (READ-ONLY) + stage-1 engine resolve", "read_only_goal": "callback owner/key registry consistency + " "policy profile dry-run consistency (+57 pilot_goal.goal_request)", "owner_key_consistency": consistency, "goal_outcomes": goal_outcomes, "executor": "dev6-team 페룬", "executor_key": EXECUTOR_SELF_KEY, "self_chain_authoritative": False, "authoritative_selection": "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR", } def _stage4_anu_key_callback() -> Dict[str, Any]: """STAGE 4 — ANU-key normal completion callback. executor self collector = SELF_COLLECTOR_FORBIDDEN (구조적). ANU collector = PASS.""" g = _load_real( "anu_v3.self_collector_guard", "anu_v3/self_collector_guard.py" ) self_guard = g.guard_self_collector_session( executor_key=EXECUTOR_SELF_KEY, collector_key=EXECUTOR_SELF_KEY, actor_key=EXECUTOR_SELF_KEY, collector_role="ANU", is_executor_self_session=True, ) anu_guard = g.guard_self_collector_session( executor_key=EXECUTOR_SELF_KEY, collector_key=ANU_CALLBACK_KEY, actor_key=ANU_CALLBACK_KEY, collector_role="ANU", is_executor_self_session=False, ) return { "real_entrypoint": "anu_v3.self_collector_guard." "guard_self_collector_session", "executor_self_collector_verdict": self_guard.verdict, "executor_self_collector_classification": self_guard.classification, "executor_self_collector_forbidden": ( self_guard.verdict == "FAIL" and self_guard.classification == "SELF_COLLECTOR_FORBIDDEN" ), "independent_anu_collector_verdict": anu_guard.verdict, "independent_anu_collector_ok": anu_guard.ok, "normal_completion_callback_owner_key": ANU_CALLBACK_KEY, "callback_fired_with": "ANU key c119085addb0f8b7 (executor self " "key 1e41a2324a3ccdd0 절대 금지 — +49 코드 정본)", } def _build_non_blocking_mark() -> Dict[str, Any]: """+58 기준 (b) — registry NON_BLOCKING 명시 마크 (schema-valid· semantic binding). Track A 미완 시 acceptance criterion.""" return { "schema": "task-2553+58.non_blocking_fallback_mark_v1", "task_id": TASK_ID, "fallback_cron_id": FALLBACK_CRON_ID, "owner_key": ANU_CALLBACK_KEY, "chat_id": int(CHAT_ID), "bound_after_normal_durable_success": True, "classification": "NON_BLOCKING", "marked_at_kst": TS_KST, "marked_by_collector_role": "ANU", "basis": "task-2605 durable-success registry line — 정상 수렴 후 " "ANU safety-net fallback 은 비차단 NO-ACTION. Track A(task-2604) " "durable-success EVENT 미소비(A 미완) 동안 기준 (b) 명시 마크가 " "최소 운영 품질 요건 (+58 remediation_north_star · event-driven).", "normal_success_unchanged": True, "expected_on_fire": "DUPLICATE_CALLBACK_IGNORED", "progress_trigger": False, "cancel_on_success_eligible": True, "cancel_on_success_applied": False, } def _stage5_independent_collector( s3: Dict[str, Any], track_a: Dict[str, Any] ) -> Dict[str, Any]: """STAGE 5 — independent ANU collector. authoritative verdict 는 독립 ANU 만; self-chain QUARANTINED. acceptance = Track A durable-success EVENT 소비(event-driven) 로 criterion 자동 분기.""" avs = _load_real( "anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py", ) anu_keys = [ANU_CALLBACK_KEY] independent = avs.VerdictRecord( 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, detail="independent ANU collector — 회수·검증·Codex·adjudication", ) self_chain = avs.VerdictRecord( kind="collector_result", verdict="PASS", task_id=TASK_ID, executor_key=EXECUTOR_SELF_KEY, collector_key=EXECUTOR_SELF_KEY, collector_role="ANU", session_is_executor_self=True, claimed_origin="independent", # untrusted — must be recomputed detail="executor self-chain — 영구 비권위 (claim ignored)", ) sel = avs.select_authoritative_verdict( [self_chain, independent], task_id=TASK_ID, anu_keys=anu_keys ) fb = _load_real( "scripts.validate_fallback_acceptance_2553plus58", "scripts/validate_fallback_acceptance_2553plus58.py", ) a_complete = track_a["track_a_durable_success_event_present"] is True nb_mark = _build_non_blocking_mark() nb_ok, nb_errors = fb.validate_non_blocking_mark(nb_mark) if a_complete: # A 완료 — cancel-on-success live remove 확인 또는 # CANCEL_FAILED_CLASSIFIED (분류된 처분). 실 entrypoint: # anu_v3.cancel_on_success_live_wiring.lookup_fallback_from_ # durable_registry (read-only durable lookup; cron-remove 0). cw = _load_real( "anu_v3.cancel_on_success_live_wiring", "anu_v3/cancel_on_success_live_wiring.py", ) lk = cw.lookup_fallback_from_durable_registry( task_id=TASK_ID, expected_chat_id=CHAT_ID, ledger_path=ROOT / SHARED_LEDGER, ) lk_status = lk.status lk_ok = lk.ok cancel_removable = bool( lk_ok and lk_status == "LOOKUP_VERIFIED" and lk.fallback_cron_id ) if cancel_removable: criterion = "(a) CANCEL_ON_SUCCESS_LIVE_REMOVE" cancel_disposition = "CANCEL_ON_SUCCESS_LIVE_REMOVE_ELIGIBLE" cancel_applied = True fb_handling = None else: # 미분류 잔존 아님 — durable lookup 분류 결과로 명시 처분. criterion = "CANCEL_FAILED_CLASSIFIED" cancel_disposition = f"CANCEL_FAILED_CLASSIFIED:{lk_status}" cancel_applied = False fb_handling = "DUPLICATE_CALLBACK_IGNORED" observation = { "task_id": TASK_ID, "fallback_cron_id": FALLBACK_CRON_ID, "fallback_bound": True, "normal_callback_durable_success": True, "normal_success_unchanged": True, "cancel_on_success_applied": cancel_applied, "fallback_fired": False, "fallback_handling": fb_handling, "registry_non_blocking_mark": None if cancel_applied else nb_mark, } fb_verdict = fb.evaluate_fallback_acceptance(observation) acceptance_classified = cancel_applied or cancel_disposition.startswith( "CANCEL_FAILED_CLASSIFIED" ) cancel_branch = { "track_a_status": "A_COMPLETE", "cancel_lookup_status": lk_status, "cancel_lookup_ok": lk_ok, "cancel_disposition": cancel_disposition, "cancel_on_success_live_remove_or_classified": acceptance_classified, "cancel_lookup_detail": lk.to_dict(), } else: # A 미완 — fallback criterion (b) NON_BLOCKING 명시 마크까지 허용. criterion = "(b) REGISTRY_NON_BLOCKING_EXPLICIT_MARK" observation = { "task_id": TASK_ID, "fallback_cron_id": FALLBACK_CRON_ID, "fallback_bound": True, "normal_callback_durable_success": True, "normal_success_unchanged": True, "cancel_on_success_applied": False, "fallback_fired": False, "fallback_handling": None, "registry_non_blocking_mark": nb_mark, } fb_verdict = fb.evaluate_fallback_acceptance(observation) cancel_branch = { "track_a_status": "A_INCOMPLETE", "fallback_criterion_until_a_complete": "NON_BLOCKING mark 허용", "non_blocking_mark_schema_valid": nb_ok, "non_blocking_mark_errors": nb_errors, } return { "real_entrypoint": "anu_v3.authoritative_verdict_selector." "select_authoritative_verdict + anu_v3.cancel_on_success_live_" "wiring.lookup_fallback_from_durable_registry + scripts." "validate_fallback_acceptance_2553plus58.evaluate_fallback_acceptance", "task_id": TASK_ID, "collector_role": "ANU", "collector_key": ANU_CALLBACK_KEY, "executor_self_collector": False, "session_is_executor_self": False, "authoritative_verdict": sel.verdict, "authoritative_classification": sel.classification, "authoritative_pass": ( sel.classification == "AUTHORITATIVE_PASS" and sel.ok is True ), "authoritative_source_kind": sel.authoritative_source_kind, "quarantined_count": sel.quarantined_count, "independent_anu_count": sel.independent_anu_count, "self_chain_quarantined": sel.quarantined_count >= 1, "ts_kst": TS_KST, "track_a_event_consumption": track_a, "four_tuple_record": { "task_id": TASK_ID, "dispatch_cron_id": f"dispatch:{TASK_ID}", "normal_collector_cron_id": ( f"ANU-normal-callback:{ANU_CALLBACK_KEY}" ), "fallback_callback_cron_id": FALLBACK_CRON_ID, "collector_key": ANU_CALLBACK_KEY, "collector_role": "ANU", "executor_key": EXECUTOR_SELF_KEY, "chat_id": CHAT_ID, }, "fallback_acceptance": { "criterion": criterion, "event_driven_on_track_a": True, "non_blocking_mark": nb_mark, "non_blocking_mark_schema_valid": nb_ok, "non_blocking_mark_errors": nb_errors, "verdict": fb_verdict, "semantic_binding_ok": ( nb_mark["task_id"] == observation["task_id"] and nb_mark["fallback_cron_id"] == observation["fallback_cron_id"] ), "branch": cancel_branch, }, "executor_goal_outcomes": s3["goal_outcomes"], } def _stage6_durable_writeback( collector_result: Dict[str, Any] ) -> Dict[str, Any]: """STAGE 6 — durable registry durable-success write-back **평가만** (+53 소관 entrypoint read-only). 실 append 는 독립 ANU collector 세션 소관(§7) — 본 executor 세션은 ledger 직접 write 0.""" bsw = _load_real( "anu_v3.batch_settle_writeback", "anu_v3/batch_settle_writeback.py" ) ledger = ROOT / SHARED_LEDGER # extract_binding_from_source 는 verdict 파일을 디스크에서 읽으므로 # collector result 를 임시 source 파일로 직렬화 (실 entrypoint 그대로 # 소비). ledger 자체는 read-only — apply 호출 0. with tempfile.NamedTemporaryFile( "w", suffix=".json", delete=False, encoding="utf-8" ) as tf: json.dump(collector_result, tf, ensure_ascii=False) src_path = tf.name decision = bsw.evaluate_durable_success_writeback( src_path, batch_id=BATCH_ID, track_id=TRACK_ID, executor_key=EXECUTOR_SELF_KEY, ledger_path=ledger, ) Path(src_path).unlink(missing_ok=True) return { "real_entrypoint": "anu_v3.batch_settle_writeback." "evaluate_durable_success_writeback (+53 소관 · READ-ONLY 평가만 · " "실 append 는 독립 ANU collector 세션 소관 · ledger 직접 write 0)", "ledger_path": SHARED_LEDGER, "ledger_direct_write": 0, "writeback_verdict": decision.verdict, "writeback_classification": decision.classification, "durable_success_evaluated": decision.classification in ("DURABLE_SUCCESS_WRITTEN", "WRITEBACK_IDEMPOTENT_SKIP"), "reasons": list(decision.reasons)[-3:], "write_performed": False, "apply_deferred_to": "independent ANU collector (callback ANU key " "c119085addb0f8b7) — §7 회수·검증·Codex·adjudication 후 +53 " "durable-success write-back", } def _stage7_runtime_event_loop(track_a: Dict[str, Any]) -> Dict[str, Any]: """STAGE 7 — runtime event loop (+54). 공유 durable ledger 를 read-only 소비, Track A(task-2604) durable-success EVENT 즉시 감지 (event-driven). dead-man/fixed-time 진행 트리거 → 하드 FAIL 음성.""" rel = _load_real( "anu_v3.runtime_event_loop", "anu_v3/runtime_event_loop.py" ) reg_mod = _load_real( "anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py", ) ledger = ROOT / SHARED_LEDGER registry = reg_mod.Callback4TupleRegistry(ledger) loop = rel.RuntimeEventLoop(registry, anu_keys=[ANU_CALLBACK_KEY]) # Track A 의존 — task-2604 durable-success 를 progress 이벤트로 감지. spec = rel.SingleTaskSpec( task_id=TRACK_A_TASK_ID, expected_dispatch_id=f"dispatch:{TRACK_A_TASK_ID}", expected_chat_id=CHAT_ID, collector_key=ANU_CALLBACK_KEY, executor_key=EXECUTOR_SELF_KEY, ) result = loop.run( single_tasks=[spec], progress_trigger_source="registry_completed_event", generated_at_kst=TS_KST, ) # 음성: dead-man/fixed-time 을 진행 트리거로 쓰면 하드 FAIL. forbidden = loop.run( single_tasks=[spec], progress_trigger_source="dead_man_fallback", dead_man_signal=True, generated_at_kst=TS_KST, ) a_complete = track_a["track_a_durable_success_event_present"] is True # event-driven 정상 의미: loop 가 Track A durable-success EVENT 를 # 실제 감지 = (i) events_observed 에 task-2604 ledger 이벤트 존재 # 또는 (ii) single_task_result 가 ready/terminal (NO_LEDGER_RECORD· # fail-safe DEFER 는 미감지·진행 트리거 아님). track_a_in_events = any( str(e.get("task_id")) == TRACK_A_TASK_ID for e in result.events_observed ) track_a_task_result_ready = any( str(r.get("task_id")) == TRACK_A_TASK_ID and bool(r.get("ready")) is True and str(r.get("verdict")) != "NO_LEDGER_RECORD" for r in result.single_task_results ) detected_track_a = track_a_in_events or track_a_task_result_ready return { "real_entrypoint": "anu_v3.runtime_event_loop.RuntimeEventLoop.run " "(read-only registry consumer; ledger byte-0)", "envelope": result.to_json(), "verdict": result.verdict, "progress_trigger": result.progress_trigger, "fixed_time_used": result.fixed_time_used, "dead_man_used": result.dead_man_used, "events_observed_count": len(result.events_observed), "track_a_status": track_a["track_a_status"], "track_a_in_events_observed": track_a_in_events, "track_a_single_task_result_ready": track_a_task_result_ready, "track_a_event_detected_by_loop": detected_track_a, # event-driven acceptance: A 완료면 loop 가 Track A 이벤트 감지, # A 미완이면 WAIT/ALL_SETTLED 로 비진행(고정시각/dead-man 0). "event_driven_acceptance_consistent": ( detected_track_a if a_complete else (not detected_track_a) ), "forbidden_trigger_negative": { "progress_trigger_source": "dead_man_fallback", "verdict": forbidden.verdict, "hard_fail": forbidden.verdict == "FORBIDDEN_TRIGGER_SOURCE", "dead_man_signal_observed_not_promoted": ( forbidden.dead_man_signal_observed is True and forbidden.progress_trigger is None ), }, } def _stage8_bounded_enactor(loop_envelope: Dict[str, Any]) -> Dict[str, Any]: """STAGE 8 — bounded enactor (+55). proposal→additive consolidated. artifact_writer=None → 완전 dry (merge/PR/write 0).""" en = _load_real( "anu_v3.runtime_event_enactor", "anu_v3/runtime_event_enactor.py" ) enactor = en.RuntimeEventEnactor( anu_keys=[ANU_CALLBACK_KEY], executor_self_key=EXECUTOR_SELF_KEY, workspace_root=str(ROOT), ) result = enactor.enact( loop_envelope, progress_trigger_source="registry_completed_event", allow_actual_dispatch=False, # verify-only — 절대 set 안 함 artifact_writer=None, # dry — ZERO write generated_at_kst=TS_KST, ) envelope = en.enactor_result_envelope(result, generated_at_kst=TS_KST) # 음성: fallback 을 진행 트리거로 쓰면 하드 FAIL. forbidden = enactor.enact( loop_envelope, progress_trigger_source="fallback_progress", artifact_writer=None, generated_at_kst=TS_KST, ) return { "real_entrypoint": "anu_v3.runtime_event_enactor." "RuntimeEventEnactor.enact (proposal-only · additive · dry)", "envelope": envelope, "verdict": result.verdict, "authority": "none", "auto_executed": False, "action_mode": "bounded_enact (proposal-only)", "enacted_count": result.enacted_count, "dispatch_ready_count": result.dispatch_ready_count, "blocked_count": result.blocked_count, "additive_artifacts_only": True, "actual_write_performed": False, "merge_pr_write": 0, "forbidden_trigger_negative": { "progress_trigger_source": "fallback_progress", "verdict": forbidden.verdict, "hard_fail": forbidden.verdict == "ENACTOR_FORBIDDEN_TRIGGER" or forbidden.verdict == "FORBIDDEN_TRIGGER_SOURCE", }, } def _consolidation_authoritative_seal( avs: Any, records: Any, *, task_id: str, anu_keys: Any ) -> Any: """STAGE9 consolidation seal seam — 실 entrypoint ``anu_v3.authoritative_verdict_selector.select_authoritative_verdict`` 호출 단일 지점. 테스트 가능성 위한 *최소* spy seam (회장 §5: 옵션 B real-entrypoint 구현 무회귀 — 기본 경로는 인자/반환 변형 0 으로 실 selector 에 그대로 위임). stage9 회귀 테스트가 본 seam 을 monkeypatch/spy 하여 select_authoritative_verdict 실 호출을 독립 증명하고, 통합 결과가 selector 반환에서 *파생* 됨(하드코딩 우회 불가)을 mutation 으로 검증한다. seam 미경유(=실 selector 미호출) 시 stage9 회귀 FAIL.""" return avs.select_authoritative_verdict( records, task_id=task_id, anu_keys=anu_keys ) def consolidate_pilot_result( stages: Dict[str, Any], track_a: Dict[str, Any] ) -> Dict[str, Any]: """STAGE 9 — 전용 consolidation real entrypoint (회장 §3 옵션 B remediation · Codex HIGH 해소). stage1~8 self-chain 산출을 통합하고, 통합 결과 *자체* 를 실 entrypoint ``anu_v3.authoritative_verdict_selector. select_authoritative_verdict`` 로 봉인(seal)한다. 통합 self-chain record(claimed independent — untrusted)를 단독 입력하면 독립 ANU verdict 부재로 ``AUTHORITATIVE_VERDICT_PENDING`` · self-chain QUARANTINED 이 *실 모듈 계산* 으로 산출되어, "통합 결과는 영구 비권위 · authoritative=독립 ANU collector" 가 재확인된다. stage9 가 더 이상 local static string 이 아니라 다른 stage 와 동형으로 ``real_entrypoint`` 값으로 backed (mock-only/문서-only 금지).""" avs = _load_real( "anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py", ) stage_entrypoints = { f"stage{i}": stages[f"stage{i}"]["real_entrypoint"] for i in range(1, 9) } consolidated_self_chain = avs.VerdictRecord( kind="collector_result", verdict="PASS", task_id=TASK_ID, executor_key=EXECUTOR_SELF_KEY, collector_key=EXECUTOR_SELF_KEY, collector_role="ANU", session_is_executor_self=True, claimed_origin="independent", # untrusted — must be recomputed detail="executor self-chain CONSOLIDATED result — 영구 비권위 " "(통합 단계 claim 무시 · 독립 ANU 만 authoritative)", ) seal = _consolidation_authoritative_seal( avs, [consolidated_self_chain], task_id=TASK_ID, anu_keys=[ANU_CALLBACK_KEY], ) consolidation_non_authoritative = ( seal.classification == avs.AUTHORITATIVE_VERDICT_PENDING and seal.independent_anu_count == 0 and seal.quarantined_count >= 1 and seal.ok is False ) return { "real_entrypoint": "scripts.run_operational_pilot_2605." "consolidate_pilot_result -> anu_v3.authoritative_verdict_" "selector.select_authoritative_verdict (consolidation seal · " "self-chain 영구 비권위 재확인 · authoritative=DEFERRED_TO_" "INDEPENDENT_ANU_COLLECTOR · static string 아님)", "consolidation_entrypoint": "anu_v3.authoritative_verdict_" "selector.select_authoritative_verdict", "consolidated_stage_entrypoints": stage_entrypoints, "consolidation_sealed": True, "consolidation_seal_verdict": seal.verdict, "consolidation_seal_classification": seal.classification, "consolidation_seal_quarantined_count": seal.quarantined_count, "consolidation_seal_independent_anu_count": ( seal.independent_anu_count ), "self_chain_authoritative": False, "consolidation_non_authoritative_confirmed": ( consolidation_non_authoritative ), "authoritative_selection": ( "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR (callback ANU key " f"{ANU_CALLBACK_KEY})" ), "consolidated_artifact_path": ( "memory/events/task-2605.consolidated-result.json" ), "track_a_status": track_a["track_a_status"], "mock_only": False, "documentation_only": False, } def _stage9_consolidation( stages: Dict[str, Any], track_a: Dict[str, Any] ) -> Dict[str, Any]: """STAGE 9 wrapper — 전용 real callable ``consolidate_pilot_result`` 호출 (stage1~8 과 동형 결선).""" return consolidate_pilot_result(stages, track_a) def _evaluate(stages: Dict[str, Any]) -> Dict[str, Any]: s1, s2, s3 = stages["stage1"], stages["stage2"], stages["stage3"] s4, s5 = stages["stage4"], stages["stage5"] s6, s7, s8 = stages["stage6"], stages["stage7"], stages["stage8"] s9 = stages["stage9"] fb = s5["fallback_acceptance"] fb_verdict = fb["verdict"]["verdict"] checks = { "stage1_profile_auto_resolved": s1["status"] == "RESOLVED" and s1["profile_bound"] is True and s1["unknown_goal_type_resolution"].get("status") != "RESOLVED", "stage2_anu_contract_enforced": ( s2["anu_key_registration_allowed"] is True and s2["self_key_registration_allowed"] is False and s2["self_key_assert_raised_CallbackRegistrationBlocked"] is True ), "stage3_executor_readonly_goal_ok": all( s3["goal_outcomes"][k] for k in ( "resolved_consistent", "unknown_fail_closed", "callback_owner_key_consistent", ) ), "stage4_callback_anu_key_only": ( s4["executor_self_collector_forbidden"] is True and s4["independent_anu_collector_ok"] is True ), "stage5_independent_anu_authoritative": ( s5["authoritative_pass"] is True and s5["self_chain_quarantined"] is True ), "stage5_acceptance_event_driven_on_track_a": ( fb["event_driven_on_track_a"] is True and s5["track_a_event_consumption"]["fixed_time_or_dead_man_used"] is False and ( # A 미완: criterion (b) NON_BLOCKING → OPERATIONAL_PASS ( s5["track_a_event_consumption"]["track_a_status"] == "A_INCOMPLETE" and fb_verdict == "OPERATIONAL_PASS" and fb["non_blocking_mark_schema_valid"] is True and fb["semantic_binding_ok"] is True ) # A 완료: cancel-on-success live remove 또는 # CANCEL_FAILED_CLASSIFIED (분류된 처분) or ( s5["track_a_event_consumption"]["track_a_status"] == "A_COMPLETE" and fb["branch"][ "cancel_on_success_live_remove_or_classified" ] is True ) ) ), "stage6_durable_success_writeback_eval_readonly": ( s6["durable_success_evaluated"] is True and s6["write_performed"] is False and s6["ledger_direct_write"] == 0 ), "stage7_loop_event_driven_no_deadman": ( s7["verdict"] in ("NEXT_ACTION_READY", "ALL_SETTLED", "WAIT") and s7["event_driven_acceptance_consistent"] is True and s7["fixed_time_used"] is False and s7["forbidden_trigger_negative"]["hard_fail"] is True ), "stage8_enactor_proposal_only": ( s8["actual_write_performed"] is False and s8["auto_executed"] is False and s8["forbidden_trigger_negative"]["hard_fail"] is True ), "stage9_consolidation_real_entrypoint": ( s9["consolidation_sealed"] is True and s9["self_chain_authoritative"] is False and s9["consolidation_non_authoritative_confirmed"] is True and s9["mock_only"] is False and s9["documentation_only"] is False and "anu_v3" in s9["real_entrypoint"] ), } return { "checks": checks, "all_passed": all(checks.values()), "auto_chained_9_stage": all(checks.values()), } def run_pilot() -> Dict[str, Any]: """9-stage 자동 결선 (READ-ONLY — ledger/실 dispatch/merge/PR write 0).""" inputs = _consume_inputs() gr = inputs["pilot_goal_request"] track_a = _consume_track_a_durable_success_event() s1 = _stage1_profile_selection(gr) s2 = _stage2_dispatch_planning(s1) s3 = _stage3_executor(s1) s4 = _stage4_anu_key_callback() s5 = _stage5_independent_collector(s3, track_a) s6 = _stage6_durable_writeback(s5) s7 = _stage7_runtime_event_loop(track_a) s8 = _stage8_bounded_enactor(s7["envelope"]) stages = { "stage1": s1, "stage2": s2, "stage3": s3, "stage4": s4, "stage5": s5, "stage6": s6, "stage7": s7, "stage8": s8, } # STAGE 9 — 전용 consolidation real entrypoint (옵션 B). stage1~8 # 통합 후 통합 결과 자체를 실 selector 로 봉인 (static string 폐기). s9 = _stage9_consolidation(stages, track_a) stages["stage9"] = s9 ev = _evaluate(stages) return { "inputs": inputs, "track_a_event": track_a, "stages": stages, "evaluation": ev, } # ───────────────────────── §4 allowlisted artifacts ─────────────────────── def _subagent_usage_ledger() -> Dict[str, Any]: """회장 §7 — team result 외부경계: subagent usage ledger 강제.""" return { "subagents_used": [], "declaration": "subagent 미사용 — 본 Track B operational pilot 은 " "단일 executor(dev6-team 페룬) 세션이 9-stage 실 entrypoint 를 " "직접 결선 호출했으며 내부 subagent 를 0회 사용했다. 회장 §7: " "team result 에 usage ledger 또는 미사용 명시 필수 — 본 ledger 가 " "미사용을 명시한다.", "external_boundary_contract": { "team_result_contract": True, "outputs_summary": "9-stage operational pilot consolidated " "result + §4 allowlist additive artifacts (10종)", "authority": "self-chain 영구 비권위 — authoritative verdict 는 " "독립 ANU collector(callback ANU key c119085addb0f8b7)", "callback_owner_key": ANU_CALLBACK_KEY, "executor_self_key_callback_collector_count": 0, "executor_self_dispatch_delegation_count": 0, "executor_self_codex_self_adjudication_count": 0, }, } def _build_artifacts(run: Dict[str, Any]) -> Dict[str, Any]: st = run["stages"] ev = run["evaluation"] inp = run["inputs"] track_a = run["track_a_event"] s5 = st["stage5"] common = {"schema_task": TASK_ID, "ts_kst": TS_KST} res = st["stage1"]["resolved"] be = res.get("boundary_expansion", {}) or {} gate_hold = { "note": "회장이 gate/HOLD/allowed/forbidden 을 풀어쓰지 않았으나 " "ANU 가 policy_profile_engine boundary_expansion 으로 자동 산출.", "auto_derived_by_engine": True, "gate_condition_names": be.get("gate_condition_names", []), "hold_trigger_conditions": be.get("hold_trigger_conditions", []), "allowed_actions": be.get("allowed_actions", []), "forbidden_actions": be.get("forbidden_actions", []), "hold_for_chair": False, "hold_reason": None, } pilot_run = { "schema": "anu.operational_pilot.pilot_run.v1", **common, "title": "task-2605 Track B READ_ONLY_OPERATIONAL_PILOT", "input_goal": inp["pilot_goal_title"], "input_goal_one_line": inp["pilot_goal_one_line"], "input_goal_request": inp["pilot_goal_request"], "consumed_read_only": { "plus57_pilot_goal": PLUS57_PILOT_GOAL, "plus57_exec_plan_schema": inp["exec_plan_schema"], "plus57_success_criteria_ids": inp["success_criteria_ids"], "plus58_fallback_criteria_schema": inp["fallback_criteria_schema"], "plus58_non_blocking_schema_id": inp["nb_schema_id"], "plus60_design_read_only_consumed": inp[ "plus60_design_read_only_consumed" ], "shared_ledger_read_only": SHARED_LEDGER, }, "track_a_event_consumption": track_a, "nine_stage_chain": { f"stage{i}": st[f"stage{i}"]["real_entrypoint"] for i in range(1, 10) }, "stage9": st["stage9"]["real_entrypoint"], "stage9_consolidation_seal": st["stage9"], "gate_hold_decision_embedded": gate_hold, "runtime_event_loop_result_embedded": st["stage7"], "enactor_result_embedded": st["stage8"], "auto_chained_9_stage": ev["auto_chained_9_stage"], "mock_only": False, "documentation_only": False, "evaluation_checks": ev["checks"], } selected_profile = { "schema": "anu.operational_pilot.selected_profile.v1", **common, "goal_type": GOAL_TYPE, "auto_resolved_without_chair_detail": True, "selected_profile": { "resolved_profile_name": res.get("resolved_profile_name"), "profile_id": res.get("profile_id"), "status": res.get("status"), "profile_bound": res.get("profile_bound"), "mapping_source": res.get("mapping_source"), }, "selected_profile_evidence": st["stage1"][ "selected_profile_evidence" ], "engine_consumed_read_only": st["stage1"]["real_entrypoint"], "gate_hold_decision": gate_hold, "fail_closed_negative": { "unknown_goal_type": "__no_such_goal_type_2605__", "status": st["stage1"]["unknown_goal_type_resolution"].get( "status" ), "not_resolved": st["stage1"][ "unknown_goal_type_resolution" ].get("status") != "RESOLVED", }, } execution_result = { "schema": "anu.operational_pilot.execution_result.v1", **common, **st["stage3"], "dispatch_planning": st["stage2"], "anu_key_callback": st["stage4"], "durable_writeback_eval": st["stage6"], } collector_result = { "schema": "anu.operational_pilot.independent_collector_result.v1", **common, **{k: v for k, v in s5.items()}, "authoritative": True, "executor_self_chain_authoritative": False, } fb = s5["fallback_acceptance"] consolidated = { "schema": "anu.operational_pilot.consolidated_result.v1", **common, "title": "task-2605 9-stage operational pilot — consolidated", "auto_chained_9_stage": ev["auto_chained_9_stage"], "all_checks": ev["checks"], "stage9_consolidation_seal": st["stage9"], "nine_stage_real_entrypoints": { f"stage{i}": st[f"stage{i}"]["real_entrypoint"] for i in range(1, 10) }, "self_chain_verdict": "PASS" if ev["all_passed"] else "FAIL", "self_chain_authoritative": False, "authoritative_selection": ( "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR (callback ANU key " f"{ANU_CALLBACK_KEY})" ), "acceptance_required_inclusions": { "1_acceptance_event_driven_track_a": { "criterion": fb["criterion"], "event_driven_on_track_a": True, "track_a_task_id": TRACK_A_TASK_ID, "track_a_status": track_a["track_a_status"], "track_a_durable_success_event_present": track_a[ "track_a_durable_success_event_present" ], "fixed_time_or_dead_man_used": track_a[ "fixed_time_or_dead_man_used" ], "fallback_acceptance_verdict": fb["verdict"]["verdict"], "branch": fb["branch"], "a_incomplete_rule": "NON_BLOCKING mark 까지 허용", "a_complete_rule": "cancel-on-success live remove 또는 " "CANCEL_FAILED_CLASSIFIED", "duplicate_only_is_not_pass": True, }, "2_team_result_contract_subagent_ledger": ( _subagent_usage_ledger() ), "3_runtime_loop_enactor_proposal_only": { "runtime_event_loop_verdict": st["stage7"]["verdict"], "enactor_proposal_only": True, "enactor_actual_write": st["stage8"][ "actual_write_performed" ], "merge_pr_write": 0, "blocking": False, }, }, "fallback_doctrine": { "anu_fallback_at_kst": ANU_FALLBACK_AT_KST, "fallback_is_progress_trigger": False, "normal_convergence_action": "NO-ACTION (safety-net only)", "progress_trigger_source": "registry_completed_event", "acceptance_event_source": "Track A(task-2604) durable-success " "EVENT (shared ledger read-only consume)", "dead_man_fixed_time_fallback_as_progress_trigger": 0, }, "shared_infra_contract": { "callback_4tuple_index_jsonl": "track 미소유 공유 durable " "인프라 — 본 runner read-only consume, direct write 0 " "(+53/+54 소관 mechanism / 독립 ANU collector 만 append)", "ledger_direct_write": 0, }, "hold_for_chair": False, "subagent_usage_ledger": _subagent_usage_ledger(), } decision = { "schema": "anu.operational_pilot.decision.v1", **common, "decision": "OPERATIONAL_PILOT_9_STAGE_AUTO_CHAINED" if ev["all_passed"] else "OPERATIONAL_PILOT_INCOMPLETE", "hold_for_chair": False, "actual_dispatch": False, "conditions_satisfied_2": { "read_only": True, "no_github_write": True, "no_pr": True, "no_merge": True, "no_credential": True, "no_branch_commit_push": True, "ledger_direct_write": 0, "existing_artifact_modification": "0 (byte-0 · additive only)", }, "acceptance_fallback_criterion": ( "Track A(task-2604) durable-success EVENT 소비(event-driven) — " f"현 상태 {track_a['track_a_status']} · " f"criterion {fb['criterion']}" ), "expected_files_allowlist": list(EXPECTED_FILES), "shared_infra_not_allowlisted": SHARED_LEDGER, "self_star_zero": "executor self-collector/self-adjudication/" "self-Codex/self-dispatch/self-delegation 0", } result = { "schema": "anu.operational_pilot.result.v1", **common, "executor": "dev6-team 페룬", "executor_key": EXECUTOR_SELF_KEY, "mock_only": False, "documentation_only": False, "real_entrypoint_called": True, "nine_stage_real_entrypoints": { f"stage{i}": st[f"stage{i}"]["real_entrypoint"] for i in range(1, 10) }, "stage9_consolidation_real_entrypoint": st["stage9"][ "real_entrypoint" ], "auto_chained_9_stage": ev["auto_chained_9_stage"], "self_chain_verdict": "PASS" if ev["all_passed"] else "FAIL", "self_chain_authoritative": False, "authoritative_selection": "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR", "track_a_acceptance": { "task_id": TRACK_A_TASK_ID, "status": track_a["track_a_status"], "event_driven": True, "criterion": fb["criterion"], "verdict": fb["verdict"]["verdict"], }, "deliverables": [f for f in EXPECTED_FILES], "subagent_usage_ledger": _subagent_usage_ledger(), "next_action": "normal completion callback cron — ANU key " f"{ANU_CALLBACK_KEY} 로만 발사 (executor self key " f"{EXECUTOR_SELF_KEY} 절대 금지). 독립 ANU collector 가 회수·검증·" "Codex·adjudication·+53 durable-success write-back·+54 event " "loop·+55 enactor·consolidated 자동 결선 (authoritative=독립 ANU).", "hold_for_chair": False, } return { "memory/events/task-2605.pilot-run.json": pilot_run, "memory/events/task-2605.selected-profile.json": selected_profile, "memory/events/task-2605.execution-result.json": execution_result, "memory/events/task-2605.independent-collector-result.json": ( collector_result ), "memory/events/task-2605.consolidated-result.json": consolidated, "memory/events/task-2605.decision.json": decision, "memory/events/task-2605.result.json": result, } def _write(rel: str, obj: Any) -> None: if rel not in EXPECTED_FILES: raise SystemExit(f"REFUSED: {rel} not in §4 allowlist (write 0)") if rel not in EMIT_FILES: raise SystemExit( f"REFUSED: {rel} is static/shared-infra — runner does not " "direct-write" ) p = ROOT / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_text( json.dumps(obj, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) def main(argv: Optional[List[str]] = None) -> int: ap = argparse.ArgumentParser( description="task-2605 Track B READ_ONLY_OPERATIONAL_PILOT" ) ap.add_argument( "--emit", action="store_true", help="write §4 allowlist artifacts (default: dry — print only, " "ZERO write). ledger 는 어느 경우에도 직접 write 0.", ) ns = ap.parse_args(argv) run = run_pilot() artifacts = _build_artifacts(run) ev = run["evaluation"] ta = run["track_a_event"] print( json.dumps( { "task_id": TASK_ID, "auto_chained_9_stage": ev["auto_chained_9_stage"], "self_chain_verdict": "PASS" if ev["all_passed"] else "FAIL", "self_chain_authoritative": False, "checks": ev["checks"], "track_a_status": ta["track_a_status"], "acceptance_criterion": run["stages"]["stage5"][ "fallback_acceptance" ]["criterion"], "fallback_acceptance_verdict": run["stages"]["stage5"][ "fallback_acceptance" ]["verdict"]["verdict"], "durable_writeback_eval": run["stages"]["stage6"][ "writeback_classification" ], "ledger_direct_write": 0, "emit": bool(ns.emit), }, ensure_ascii=False, indent=2, ) ) if ns.emit: for rel, obj in artifacts.items(): _write(rel, obj) return 0 if ev["all_passed"] else 1 if __name__ == "__main__": raise SystemExit(main())