# -*- coding: utf-8 -*- """task-2553+60 — LIVE READ-ONLY OPERATIONAL PILOT (9-stage 자동 결선). 회장 §1 verbatim: 짧은 read-only goal 하나만으로 ANU 가 profile selection → dispatch planning → executor → ANU-key normal callback → independent collector → durable registry → runtime event loop → bounded enactor → consolidated result 까지 **자동 수행**하는지 검증한다. 회장이 gate/HOLD/allowed/forbidden 을 풀어쓰지 않아도 ANU 가 policy_profile_engine + runtime guard 로 자동 산출·검증·통합한다. 핵심 = 본 스크립트가 각 단계의 **실 entrypoint 를 직접 호출**(mock-only FAIL · 문서-only 금지)해 9-stage 자동 진행을 입증한다. 입력은 +57 의 짧은 read-only goal(`pilot_goal.json` goal_request, goal_type = ``runtime_structure_smoke_pilot``) 이며, +58 fallback acceptance criterion/non_blocking schema 를 read-only 로 소비한다. 실 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 소비해 profile /gate/HOLD/allowed/forbidden 자동 산출 (engine byte-0). 2. ``anu_v3.callback_owner_validator.validate_callback_owner_runtime`` + ``assert_registration_permitted`` — dispatch planning 의 ANU-key callback contract 강제 (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 completion callback owner 는 독립 ANU key 만; executor self key collector 는 ``SELF_COLLECTOR_FORBIDDEN``. 5. ``anu_v3.authoritative_verdict_selector.select_authoritative_verdict`` — independent-ANU verdict 만 authoritative, self-chain QUARANTINED. ``scripts.validate_fallback_acceptance_2553plus58.evaluate_fallback_ acceptance`` — +58 기준(b) NON_BLOCKING schema-valid·semantic binding. 6. ``anu_v3.batch_settle_writeback.evaluate_durable_success_writeback`` + ``apply_durable_success_writeback`` — +44 ledger 에 durable-success additive append-only (idempotent; +53 소관). 7. ``anu_v3.runtime_event_loop.RuntimeEventLoop.run`` — durable-registry append 를 progress 이벤트로 즉시 감지 (registry_completed_event 만; dead-man/fixed-time/fallback 진행 트리거 → 하드 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·subagent usage 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 기존 산출물 변조. 본 스크립트의 유일한 durable write 는 (i) §6 allowlist 의 task-2553+60 신규 산출물 (``--emit`` 시에만) 과 (ii) +53 소관 ``apply_durable_success_writeback`` 가 +44 ledger 에 수행하는 durable-success additive append(append-only, idempotent) 뿐이다. enactor 는 artifact_writer=None 으로 완전 dry. executor(dev7 이참나, key a999e2ea4c06d2fb) 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-2553+60" GOAL_TYPE = "runtime_structure_smoke_pilot" BATCH_ID = "task-2553+60-operational-pilot" TRACK_ID = "OPERATIONAL_PILOT" # 회장 §9 verbatim — 독립 ANU collector key. executor self key 절대 금지. ANU_CALLBACK_KEY = "c119085addb0f8b7" EXECUTOR_SELF_KEY = "a999e2ea4c06d2fb" # dev7 이참나 (발사 금지 대상) CHAT_ID = "6937032012" TS_KST = "2026-05-18 22:40 KST" # 본 배치 ANU fallback — 안전망 한정 (진행/수락 트리거 아님). +57/+58 정합. ANU_FALLBACK_AT_KST = "2026-05-19 01:05 KST" FALLBACK_CRON_ID = f"ANU-fallback-safety-net:{TASK_ID}:{ANU_CALLBACK_KEY}" # 선행 read-only consume (byte-0). +57 짧은 goal·+58 fallback 기준. 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" # §6 expected_files allowlist (이 외 write 0 — additive only). EXPECTED_FILES = ( "memory/events/task-2553+60.pilot-run.json", "memory/events/task-2553+60.selected-profile.json", "memory/events/task-2553+60.gate-hold-decision.json", "memory/events/task-2553+60.execution-result.json", "memory/events/task-2553+60.independent-collector-result.json", "memory/events/task-2553+60.runtime-event-loop-result.json", "memory/events/task-2553+60.enactor-result.json", "memory/events/task-2553+60.consolidated-result.json", "scripts/run_operational_pilot_2553plus60.py", "tests/regression/test_operational_pilot_2553plus60.py", "memory/events/task-2553+60.decision.json", "memory/events/task-2553+60.result.json", "memory/reports/task-2553+60.md", "memory/events/callback_4tuple_index.jsonl", ) # 본 runner 가 --emit 시 직접 write 하는 산출물. 스크립트/테스트/리포트는 # 정적(byte-0 자기보존), ledger 는 +53 apply_durable_success_writeback # 경유(직접 _write 금지)이므로 EMIT_FILES 에서 제외. EMIT_FILES = ( "memory/events/task-2553+60.pilot-run.json", "memory/events/task-2553+60.selected-profile.json", "memory/events/task-2553+60.gate-hold-decision.json", "memory/events/task-2553+60.execution-result.json", "memory/events/task-2553+60.independent-collector-result.json", "memory/events/task-2553+60.runtime-event-loop-result.json", "memory/events/task-2553+60.enactor-result.json", "memory/events/task-2553+60.consolidated-result.json", "memory/events/task-2553+60.decision.json", "memory/events/task-2553+60.result.json", ) class MockOnlyError(RuntimeError): """실 entrypoint 미사용(mock-only) 시 즉시 FAIL — 문서-only 금지.""" def _load_real(modname: str, relpath: str): """정본 모듈을 파일 경로로 로딩 (mock 0). repo 에 dispatch.py 와 dispatch/ 가 공존해 일반 import 가 컨텍스트에 따라 엇갈리므로 sibling +51/+57 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 (sibling # +51 pilot 동형 — collision-proof, mock 0). _PRESEED = ( ("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.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.""" pg = _read_json(PLUS57_PILOT_GOAL) goal_request = pg["pilot_goal"]["goal_request"] return { "pilot_goal_request": goal_request, "pilot_goal_title": pg["pilot_goal"]["title"], "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"], } # ───────────────────────── 9-stage real-entrypoint chain ─────────────────── def _stage1_profile_selection(goal_request: Dict[str, Any]) -> Dict[str, Any]: """STAGE 1 — ANU profile selection. 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_2553plus60__", "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 강제. 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 — 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": "normal-callback durable-success event", }, "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": "policy profile engine read-only resolve " "consistency dry-run (+57 pilot_goal.goal_request)", "owner_key_consistency": consistency, "goal_outcomes": goal_outcomes, "executor": "dev7-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 a999e2ea4c06d2fb 절대 금지 — +49 코드 정본)", } def _build_non_blocking_mark() -> Dict[str, Any]: """+58 기준 (b) — registry NON_BLOCKING 명시 마크 (schema-valid· semantic binding). normal durable-success 이후 잔존 bound ANU safety-net fallback 의 비차단 명시. cancel_on_success_applied=false.""" 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-2553+60 durable-success registry line (+53 " "apply_durable_success_writeback) — 정상 수렴 후 ANU safety-net " "fallback 은 비차단 NO-ACTION. cancel-on-success live adoption " "전까지 기준 (b) 명시 마크가 최소 운영 품질 요건 (+58 " "remediation_north_star).", "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]) -> Dict[str, Any]: """STAGE 5 — independent ANU collector. authoritative verdict 는 독립 ANU 만; self-chain QUARANTINED. +58 fallback acceptance 판정.""" 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 ) # +58 fallback acceptance — 기준 (b) NON_BLOCKING 명시 마크. nb_mark = _build_non_blocking_mark() fb = _load_real( "scripts.validate_fallback_acceptance_2553plus58", "scripts/validate_fallback_acceptance_2553plus58.py", ) nb_ok, nb_errors = fb.validate_non_blocking_mark(nb_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) return { "real_entrypoint": "anu_v3.authoritative_verdict_selector." "select_authoritative_verdict + 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, "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": "(b) REGISTRY_NON_BLOCKING_EXPLICIT_MARK", "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"] ), }, "executor_goal_outcomes": s3["goal_outcomes"], } def _stage6_durable_writeback( collector_result: Dict[str, Any], *, do_write: bool ) -> Dict[str, Any]: """STAGE 6 — durable registry write-back (+53 소관). +44 ledger 에 durable-success additive append-only (idempotent).""" bsw = _load_real( "anu_v3.batch_settle_writeback", "anu_v3/batch_settle_writeback.py" ) ledger = ROOT / "memory/events/callback_4tuple_index.jsonl" # extract_binding_from_source 는 verdict 파일을 디스크에서 읽으므로 # collector result 를 source 파일로 직렬화 (실 entrypoint 그대로 소비). 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, ) applied = None if do_write: applied = bsw.apply_durable_success_writeback( decision, ledger_path=ledger ) Path(src_path).unlink(missing_ok=True) final = applied or decision return { "real_entrypoint": "anu_v3.batch_settle_writeback." "evaluate_durable_success_writeback + apply_durable_success_writeback " "(+53 소관 · +44 ledger additive append-only · idempotent)", "ledger_path": "memory/events/callback_4tuple_index.jsonl", "writeback_verdict": final.verdict, "writeback_classification": final.classification, "appended": bool(getattr(final, "appended", False)), "durable_success": final.classification in ("DURABLE_SUCCESS_WRITTEN", "WRITEBACK_IDEMPOTENT_SKIP"), "reasons": list(final.reasons)[-3:], "write_performed": bool(do_write), } def _stage7_runtime_event_loop() -> Dict[str, Any]: """STAGE 7 — runtime event loop (+54). durable-registry append 를 progress 이벤트로 즉시 감지. dead-man/fixed-time 진행 트리거 → 하드 FAIL 음성 입증 (진행 트리거는 registry_completed_event 만).""" 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 / "memory/events/callback_4tuple_index.jsonl" registry = reg_mod.Callback4TupleRegistry(ledger) loop = rel.RuntimeEventLoop(registry, anu_keys=[ANU_CALLBACK_KEY]) spec = rel.SingleTaskSpec( task_id=TASK_ID, expected_dispatch_id=f"dispatch:{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, ) return { "real_entrypoint": "anu_v3.runtime_event_loop.RuntimeEventLoop.run " "(read-only registry consumer; registry 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), "detected_durable_success": any( str(e.get("task_id")) == TASK_ID for e in result.events_observed ) or any( str(r.get("task_id")) == TASK_ID for r in result.single_task_results ), "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 _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"] 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 and s5["fallback_acceptance"]["verdict"]["verdict"] == "OPERATIONAL_PASS" and s5["fallback_acceptance"]["non_blocking_mark_schema_valid"] is True and s5["fallback_acceptance"]["semantic_binding_ok"] is True ), "stage6_durable_success_writeback": s6["durable_success"] is True, "stage7_loop_detected_event": ( s7["verdict"] in ("NEXT_ACTION_READY", "ALL_SETTLED", "WAIT") and s7["detected_durable_success"] is True 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 ), } return { "checks": checks, "all_passed": all(checks.values()), "auto_chained_9_stage": all(checks.values()), } def run_pilot(*, do_write: bool) -> Dict[str, Any]: inputs = _consume_inputs() gr = inputs["pilot_goal_request"] s1 = _stage1_profile_selection(gr) s2 = _stage2_dispatch_planning(s1) s3 = _stage3_executor(s1) s4 = _stage4_anu_key_callback() s5 = _stage5_independent_collector(s3) s6 = _stage6_durable_writeback(s5, do_write=do_write) s7 = _stage7_runtime_event_loop() s8 = _stage8_bounded_enactor(s7["envelope"]) stages = { "stage1": s1, "stage2": s2, "stage3": s3, "stage4": s4, "stage5": s5, "stage6": s6, "stage7": s7, "stage8": s8, } ev = _evaluate(stages) return {"inputs": inputs, "stages": stages, "evaluation": ev} # ───────────────────────── §6 allowlisted artifacts ─────────────────────── def _subagent_usage_ledger() -> Dict[str, Any]: """회장 §2/§4(2) — team result 외부경계: subagent usage ledger 강제.""" return { "subagents_used": [], "declaration": "subagent 미사용 — 본 operational pilot 은 단일 " "executor(dev7 이참나) 세션이 9-stage 실 entrypoint 를 직접 결선 " "호출했으며 내부 subagent 를 0회 사용했다. 회장 §2: 팀장 내부 " "subagent 사용은 재량이나 team result 에 usage ledger 또는 미사용 " "명시 필수 — 본 ledger 가 미사용을 명시한다.", "external_boundary_contract": { "team_result_contract": True, "outputs_summary": "9-stage operational pilot consolidated " "result + §6 allowlist additive artifacts", "authority": "self-chain 영구 비권위 — authoritative verdict 는 " "독립 ANU collector(callback ANU key c119085addb0f8b7)", "callback_owner_key": ANU_CALLBACK_KEY, "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"] s5 = st["stage5"] common = {"schema_task": TASK_ID, "ts_kst": TS_KST} pilot_run = { "schema": "anu.operational_pilot.pilot_run.v1", **common, "title": "task-2553+60 LIVE READ-ONLY OPERATIONAL PILOT", "input_goal": inp["pilot_goal_title"], "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"], }, "nine_stage_chain": { f"stage{i}": st[f"stage{i}"]["real_entrypoint"] for i in range(1, 9) }, "stage9": "consolidated result (memory/events/" "task-2553+60.consolidated-result.json)", "auto_chained_9_stage": ev["auto_chained_9_stage"], "mock_only": False, "documentation_only": False, "evaluation_checks": ev["checks"], } res = st["stage1"]["resolved"] be = res.get("boundary_expansion", {}) or {} 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"], "fail_closed_negative": { "unknown_goal_type": "__no_such_goal_type_2553plus60__", "status": st["stage1"]["unknown_goal_type_resolution"].get( "status" ), "not_resolved": st["stage1"][ "unknown_goal_type_resolution" ].get("status") != "RESOLVED", }, } gate_hold = { "schema": "anu.operational_pilot.gate_hold_decision.v1", **common, "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, } execution_result = { "schema": "anu.operational_pilot.execution_result.v1", **common, **st["stage3"], "dispatch_planning": st["stage2"], "anu_key_callback": st["stage4"], } 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, } loop_result = { "schema": "anu.operational_pilot.runtime_event_loop_result.v1", **common, **st["stage7"], "durable_writeback": st["stage6"], } enactor_result = { "schema": "anu.operational_pilot.enactor_result.v1", **common, **st["stage8"], } fb = s5["fallback_acceptance"] consolidated = { "schema": "anu.operational_pilot.consolidated_result.v1", **common, "title": "task-2553+60 9-stage operational pilot — consolidated", "auto_chained_9_stage": ev["auto_chained_9_stage"], "all_checks": ev["checks"], "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_fallback_after_durable_success": { "criterion": "(b) REGISTRY_NON_BLOCKING_EXPLICIT_MARK", "fallback_acceptance_verdict": fb["verdict"]["verdict"], "non_blocking_mark_schema_valid": fb[ "non_blocking_mark_schema_valid" ], "semantic_binding_ok": fb["semantic_binding_ok"], "duplicate_only_is_not_pass": True, "note": "정상 durable-success 후 잔존 bound ANU " "safety-net fallback 을 +58 schema-valid·binding 일치 " "NON_BLOCKING 으로 명시 마크 → OPERATIONAL_PASS (DUPLICATE-" "only 아님).", }, "2_team_result_contract_subagent_ledger": ( _subagent_usage_ledger() ), "3_plus58_codex_microfix_backlog": { "tracking": "next pilot backlog (차단성 아님 — 추적만)", "items": [ { "severity": "HIGH", "ref": "+58 Codex HIGH proposal-only micro-fix #1", "disposition": "next_pilot_backlog", }, { "severity": "MEDIUM", "ref": "+58 Codex MEDIUM proposal-only micro-fix #2", "disposition": "next_pilot_backlog", }, { "severity": "MEDIUM", "ref": "+58 Codex MEDIUM proposal-only micro-fix #3", "disposition": "next_pilot_backlog", }, ], "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", "dead_man_fixed_time_fallback_as_progress_trigger": 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, "existing_artifact_modification": "0 (byte-0 · additive only)", }, "expected_files_allowlist": list(EXPECTED_FILES), "self_star_zero": "executor self-collector/self-adjudication/" "self-Codex/self-dispatch/self-delegation 0", } result = { "schema": "anu.operational_pilot.result.v1", **common, "executor": "dev7-team 이참나", "executor_key": EXECUTOR_SELF_KEY, "mock_only": False, "documentation_only": False, "real_entrypoint_called": True, "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", "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} 로만 발사 (durable-success event). 독립 ANU " "collector 가 회수·검증·Codex·adjudication·batch 통합.", "hold_for_chair": False, } return { "memory/events/task-2553+60.pilot-run.json": pilot_run, "memory/events/task-2553+60.selected-profile.json": selected_profile, "memory/events/task-2553+60.gate-hold-decision.json": gate_hold, "memory/events/task-2553+60.execution-result.json": execution_result, "memory/events/task-2553+60.independent-collector-result.json": ( collector_result ), "memory/events/task-2553+60.runtime-event-loop-result.json": ( loop_result ), "memory/events/task-2553+60.enactor-result.json": enactor_result, "memory/events/task-2553+60.consolidated-result.json": consolidated, "memory/events/task-2553+60.decision.json": decision, "memory/events/task-2553+60.result.json": result, } def _write(rel: str, obj: Any) -> None: if rel not in EXPECTED_FILES: raise SystemExit(f"REFUSED: {rel} not in §6 allowlist (write 0)") if rel not in EMIT_FILES: raise SystemExit( f"REFUSED: {rel} is static/ledger — 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-2553+60 LIVE READ-ONLY OPERATIONAL PILOT" ) ap.add_argument( "--emit", action="store_true", help="write §6 allowlist artifacts + perform +53 durable-success " "additive append (default: dry — print only, ZERO write)", ) ns = ap.parse_args(argv) run = run_pilot(do_write=ns.emit) artifacts = _build_artifacts(run) ev = run["evaluation"] 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"], "fallback_acceptance_verdict": run["stages"]["stage5"][ "fallback_acceptance" ]["verdict"]["verdict"], "durable_writeback": run["stages"]["stage6"][ "writeback_classification" ], "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())