# -*- coding: utf-8 -*- """anu_v3.runtime_event_enactor — bounded enactor for the +54 runtime-event loop proposal candidates (task-2553+55, 회장 결정). 회장 §1 verbatim: "자동 loop 가 proposal 생성까지 완성됐고, 다음 단계는 proposal candidate 를 안전 조건 안에서 enact 하는 bounded enactor." The +54 ``RuntimeEventLoop`` interprets the durable registry as progress events and emits *proposal candidates only* (zero authority). This module is the bounded enactor that consumes those candidates and, within strict safety conditions, enacts ONLY what it is authorized to (회장 §2 1~10): 1. input = a +54 runtime-event-loop result (envelope or inner result). 2. type = consolidated_summary · closeout_candidate · next_dispatch_candidate · hold_packet (``anu_v3.proposal_authorization_gate``). 3. enact = ONLY authorized actions. 4. else = the proposal is retained as a proposal (no enact). 5. closeout = an ADDITIVE artifact ONLY (a brand-new file; zero modification of any existing artifact; zero merge/PR/branch). 6. next dispatch = candidate->actual dispatch promotion ONLY on the §2.6 triple AND an explicit chair grant (verify-only here — Track B never promotes; 회장 §3.B/§8/§10). 7. merge / PR / credential / branch-write = BLOCKED by default. 8. every enact decision is recorded in an enactor-result JSON. 9. re-processing the same proposal is idempotent (``anu_v3.enactor_idempotency``). 10. using a dead-man / fixed-time / fallback source as the progress trigger is a hard FAIL. Track boundary (회장 §3): * Track A (final consolidated closeout) -> actually ENACTED here, as an ADDITIVE artifact only. * Track B (next_dispatch_candidate safety gate) -> verified ONLY. * Track C (hold_packet routing) -> verified ONLY. Authority boundary: the enactor performs ZERO merge / PR / branch / main write / credential op / cron register / dispatch / subprocess / cokacdir. The ONLY side effect it is permitted is an *additive* artifact create, and even that is delegated to an injected, allowlist-gated ``artifact_writer`` (the module never opens a file for write itself — pure core + injected sink, fully testable with a tmp writer). Layer A / NO-CRON. """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from typing import Callable, Dict, List, Optional, Sequence from anu_v3.enactor_idempotency import ( # pyright: ignore[reportMissingImports] # noqa: E501 IDEMPOTENT_BINDING_CONFLICT, IDEMPOTENT_SKIP, already_enacted, enact_id, ) from anu_v3.proposal_authorization_gate import ( # pyright: ignore[reportMissingImports] # noqa: E501 ACTION_BLOCKED, ACTION_DISPATCH_READY, ACTION_ENACT_ADDITIVE, ACTION_HOLD_ROUTED, PROPOSAL_CLOSEOUT_CANDIDATE, PROPOSAL_CONSOLIDATED_SUMMARY, authorize_proposal, ) ENACTOR_SCHEMA = "anu_v3.runtime_event_enactor.v1" RESULT_SCHEMA = "runtime_event_enactor_result.v1" CLOSEOUT_ARTIFACT_SCHEMA = "task-2553.final-consolidated-closeout.v1" # The ONLY progress trigger the enactor accepts (회장 §2.10 / §6). TRIGGER_REGISTRY_COMPLETED = "registry_completed_event" TRIGGER_FIXED_TIME = "fixed_time_gate" TRIGGER_DEAD_MAN = "dead_man_fallback" TRIGGER_FALLBACK = "fallback_progress" # Superset of the +54/+47 forbidden set: §2.10 verbatim forbids # dead-man / fixed-time / fallback as a *progress trigger*. FORBIDDEN_PROGRESS_TRIGGERS = frozenset( {TRIGGER_FIXED_TIME, TRIGGER_DEAD_MAN, TRIGGER_FALLBACK} ) # Track A additive artifact target (회장 §4 expected_files — additive only). CLOSEOUT_ARTIFACT_TARGET = ( "memory/events/task-2553.final-consolidated-closeout_260518.json" ) AUTHORITY_NONE = "none" # Enactor verdicts. ENACTOR_ENACTED = "ENACTED" ENACTOR_IDEMPOTENT_SKIP = "IDEMPOTENT_SKIP" ENACTOR_NO_ACTION = "NO_ACTION" ENACTOR_FORBIDDEN_TRIGGER = "FORBIDDEN_TRIGGER_SOURCE" ENACTOR_HOLD_FOR_CHAIR = "HOLD_FOR_CHAIR" # An injected allowlist-gated additive writer: (rel_path, payload) -> None. ArtifactWriter = Callable[[str, Dict[str, object]], None] def _sha(value: object) -> str: return hashlib.sha256( json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") ).hexdigest() def _unwrap_loop_result(loop_result: Dict[str, object]) -> Dict[str, object]: """Accept either the +54 envelope (``...runtime-event-loop-result.v1`` with a nested ``result``) or the inner ``runtime_event_loop_result.v1``. """ inner = loop_result.get("result") if isinstance(inner, dict) and inner.get("schema") == ( "runtime_event_loop_result.v1" ): return inner return loop_result @dataclass class EnactionRecord: proposal_type: str action: str source_id: str authorized: bool enacted: bool idempotent_skip: bool artifact_target: Optional[str] enact_id: Optional[str] decision: Dict[str, object] idempotency: Optional[Dict[str, object]] reasons: List[str] = field(default_factory=list) def to_json(self) -> Dict[str, object]: return { "proposal_type": self.proposal_type, "action": self.action, "source_id": self.source_id, "authorized": self.authorized, "enacted": self.enacted, "idempotent_skip": self.idempotent_skip, "artifact_target": self.artifact_target, "enact_id": self.enact_id, "decision": self.decision, "idempotency": self.idempotency, "reasons": list(self.reasons), } @dataclass class EnactorResult: schema: str generated_at_kst: str verdict: str progress_trigger: Optional[str] fixed_time_used: bool dead_man_used: bool fallback_used: bool authority: str loop_result_sha256: str proposals_total: int enacted_count: int idempotent_skip_count: int dispatch_ready_count: int blocked_count: int hold_routed_count: int records: List[Dict[str, object]] additive_artifacts: List[Dict[str, object]] hold_for_chair: bool reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict not in ( ENACTOR_FORBIDDEN_TRIGGER, ENACTOR_HOLD_FOR_CHAIR, ) def to_json(self) -> Dict[str, object]: return { "schema": self.schema, "generated_at_kst": self.generated_at_kst, "verdict": self.verdict, "progress_trigger": self.progress_trigger, "fixed_time_used": self.fixed_time_used, "dead_man_used": self.dead_man_used, "fallback_used": self.fallback_used, "authority": self.authority, "auto_executed": False, "merge_pr_branch_credential": "blocked", "loop_result_sha256": self.loop_result_sha256, "proposals_total": self.proposals_total, "enacted_count": self.enacted_count, "idempotent_skip_count": self.idempotent_skip_count, "dispatch_ready_count": self.dispatch_ready_count, "blocked_count": self.blocked_count, "hold_routed_count": self.hold_routed_count, "records": list(self.records), "additive_artifacts": list(self.additive_artifacts), "hold_for_chair": self.hold_for_chair, "reasons": list(self.reasons), } class RuntimeEventEnactor: """Bounded enactor over +54 proposal candidates (read-mostly; the only side effect is an injected, allowlist-gated *additive* artifact write). """ def __init__( self, *, anu_keys: Sequence[str], executor_self_key: str = "", workspace_root: str = "/home/jay/workspace", ) -> None: self.anu_keys = tuple(k for k in anu_keys if k) self.executor_self_key = executor_self_key self.workspace_root = workspace_root # -- candidate extraction ----------------------------------------- def _collect_candidates( self, inner: Dict[str, object] ) -> List[Dict[str, object]]: out: List[Dict[str, object]] = [] for key in ( "consolidated_summary_candidates", "dispatch_candidates", "quarantined_events", ): seq = inner.get(key) if isinstance(seq, list): for item in seq: if isinstance(item, dict): out.append(item) return out def _build_closeout_artifact( self, *, eid: str, generated_at_kst: str, summary: Dict[str, object], loop_sha: str, ) -> Dict[str, object]: """Construct the ADDITIVE final-consolidated-closeout artifact. It is a brand-new, self-describing file: it carries the ``enact_id`` (for idempotency), the consolidated batch identity, the source loop sha, and an explicit authority boundary. It NEVER references or mutates any existing artifact (회장 §2.5/§3.A/§8).""" return { "schema": CLOSEOUT_ARTIFACT_SCHEMA, "enact_id": eid, "generated_at_kst": generated_at_kst, "authority": AUTHORITY_NONE, "action_mode": "enacted_additive_artifact_only", "auto_executed": False, "merge_pr_branch_credential": "blocked", "task": "task-2553", "produced_by": "task-2553+55 bounded runtime-event enactor", "source_proposal": { "schema": summary.get("schema"), "status": summary.get("status"), "batch_id": summary.get("batch_id"), "all_settled": summary.get("all_settled"), "all_authoritative_pass": summary.get( "all_authoritative_pass" ), "generated_basis": summary.get("generated_basis"), }, "source_loop_result_sha256": loop_sha, "consolidated_track_states": summary.get("track_states") or [], "closeout": { "kind": "FINAL_CONSOLIDATED_CLOSEOUT", "status": "CLOSED_OUT_ADDITIVE", "basis": ( "+54 runtime-event-loop consolidated_summary_candidate " "(all-settled & all-AUTHORITATIVE_PASS) enacted by the " "+55 bounded enactor as an additive artifact only." ), "merge": False, "pr": False, "branch_write": False, "credential_op": False, "dispatch": False, }, "note": ( "ADDITIVE ARTIFACT ONLY — this is a newly created closeout " "record. Zero existing artifact was modified; zero merge / " "PR / branch / main write / credential op / dispatch was " "performed (회장 §2.5/§3.A/§6/§8)." ), } # -- the bounded enact loop --------------------------------------- def enact( self, loop_result: Dict[str, object], *, progress_trigger_source: str = TRIGGER_REGISTRY_COMPLETED, allow_actual_dispatch: bool = False, artifact_writer: Optional[ArtifactWriter] = None, closeout_target: str = CLOSEOUT_ARTIFACT_TARGET, generated_at_kst: str = "", ) -> EnactorResult: """Consume +54 proposal candidates and enact only the authorized bounded actions (회장 §2 1~10). ``artifact_writer`` (optional) is the injected allowlist-gated additive sink. When omitted, the enactor still computes every decision + the would-be artifact + idempotency, but performs ZERO write (dry, fully deterministic). Track A actual enact requires the writer to be supplied by the allowlisted runner. """ reasons: List[str] = [] loop_sha = _sha(loop_result) # §2.10 / §6 — a dead-man / fixed-time / fallback progress trigger # is a hard FAIL. This is checked FIRST, before any classification. if progress_trigger_source in FORBIDDEN_PROGRESS_TRIGGERS: reasons.append( f"progress_trigger_source={progress_trigger_source!r} is a " "dead-man / fixed-time / fallback class trigger — FORBIDDEN " "as a progress trigger; the bounded enactor refuses to " f"enact. Only {TRIGGER_REGISTRY_COMPLETED!r} (a durable-" "registry append/update consumed by +54) drives progress " "(회장 §2.10/§6, reg 6)." ) return EnactorResult( schema=RESULT_SCHEMA, generated_at_kst=generated_at_kst, verdict=ENACTOR_FORBIDDEN_TRIGGER, progress_trigger=None, fixed_time_used=( progress_trigger_source == TRIGGER_FIXED_TIME ), dead_man_used=( progress_trigger_source == TRIGGER_DEAD_MAN ), fallback_used=( progress_trigger_source == TRIGGER_FALLBACK ), authority=AUTHORITY_NONE, loop_result_sha256=loop_sha, proposals_total=0, enacted_count=0, idempotent_skip_count=0, dispatch_ready_count=0, blocked_count=0, hold_routed_count=0, records=[], additive_artifacts=[], hold_for_chair=False, reasons=reasons, ) inner = _unwrap_loop_result(loop_result) candidates = self._collect_candidates(inner) records: List[EnactionRecord] = [] additive_artifacts: List[Dict[str, object]] = [] enacted = idem_skip = dispatch_ready = blocked = hold_routed = 0 hold = False # In-run dedupe: the consolidated_summary AND the closeout dispatch # candidate of the same batch resolve to ONE additive closeout. The # second occurrence in the same run is an idempotent skip — never a # second write, never a binding conflict (회장 §2.9). seen_enact_ids: set[str] = set() for cand in candidates: decision = authorize_proposal( cand, anu_keys=self.anu_keys, executor_self_key=self.executor_self_key, allow_actual_dispatch=allow_actual_dispatch, ) dj = decision.to_json() rec = EnactionRecord( proposal_type=decision.proposal_type, action=decision.action, source_id=decision.source_id, authorized=decision.authorized, enacted=False, idempotent_skip=False, artifact_target=None, enact_id=None, decision=dj, idempotency=None, reasons=list(decision.reasons), ) if decision.action == ACTION_ENACT_ADDITIVE and ( decision.proposal_type in ( PROPOSAL_CONSOLIDATED_SUMMARY, PROPOSAL_CLOSEOUT_CANDIDATE, ) ): # Canonical closeout identity, derived consistently from # EITHER candidate shape (the consolidated_summary carries # batch_id/all_authoritative_pass at top level; the closeout # dispatch candidate carries them as source_id/gate_evidence). ge = cand.get("gate_evidence") ge = ge if isinstance(ge, dict) else {} batch_key = str( cand.get("batch_id") or cand.get("source_id") or "" ) all_ap = bool(cand.get("all_authoritative_pass")) or bool( ge.get("all_authoritative_pass") ) # The enact_id is keyed on the STABLE logical closeout # identity (batch + all-AUTHORITATIVE_PASS), NOT the differing # proposal_type/track shape, so the consolidated_summary and # the closeout dispatch candidate of one batch resolve to the # SAME id -> one additive write, then idempotent (회장 §2.5/ # §2.9). eid = enact_id( proposal_type="final_consolidated_closeout", source_id=batch_key, settle_identity=[batch_key, all_ap], artifact_target=closeout_target, ) rec.enact_id = eid rec.artifact_target = closeout_target if eid in seen_enact_ids: rec.idempotent_skip = True idem_skip += 1 rec.idempotency = { "schema": "anu_v3.enactor_idempotency.v1", "enact_id": eid, "artifact_target": closeout_target, "classification": "IDEMPOTENT_SKIP", "should_write": False, "existing_enact_id": eid, "reasons": [ "same closeout already enacted earlier in THIS " "run (consolidated_summary + closeout candidate " "of one batch) — idempotent skip, no second " "write (회장 §2.9, reg 2)." ], } rec.reasons.append( "in-run duplicate of the same final consolidated " "closeout — idempotent skip (회장 §2.9, reg 2)." ) records.append(rec) continue seen_enact_ids.add(eid) idem = already_enacted( eid=eid, artifact_target=closeout_target, workspace_root=self.workspace_root, ) rec.idempotency = idem.to_json() if idem.classification == IDEMPOTENT_BINDING_CONFLICT: hold = True rec.reasons.append( "additive closeout path already bound to a " "DIFFERENT enact_id — never overwritten; escalated " "to HOLD (회장 §7 registry/binding mismatch)." ) elif idem.classification == IDEMPOTENT_SKIP: rec.idempotent_skip = True idem_skip += 1 rec.reasons.append( "same proposal re-processed — idempotent skip; " "additive artifact left byte-0 (회장 §2.9, reg 2)." ) else: # FIRST_WRITE artifact = self._build_closeout_artifact( eid=eid, generated_at_kst=generated_at_kst, summary=cand, loop_sha=loop_sha, ) if artifact_writer is not None: artifact_writer(closeout_target, artifact) rec.enacted = True enacted += 1 rec.reasons.append( "Track A — final consolidated closeout ENACTED " "as an ADDITIVE artifact (new file; zero " "modification of any existing artifact; zero " "merge/PR/branch; 회장 §2.5/§3.A/§8, reg 1/7)." ) else: rec.reasons.append( "Track A authorized & deterministic artifact " "computed, but no allowlisted artifact_writer " "injected — DRY (zero write). The allowlisted " "runner supplies the writer for the real enact." ) additive_artifacts.append( { "target": closeout_target, "enact_id": eid, "written": rec.enacted, "artifact": artifact if not rec.idempotent_skip else None, } ) elif decision.action == ACTION_DISPATCH_READY: dispatch_ready += 1 rec.reasons.append( "Track B — §2.6 triple PASSED -> DISPATCH-READY; " "verify-only, NO actual dispatch executed (회장 §3.B/" "§8/§10, reg 4)." ) elif decision.action == ACTION_BLOCKED: blocked += 1 rec.reasons.append( "merge/PR/credential/branch target -> BLOCKED; retained " "as a proposal (회장 §2.7/§6, reg 5)." ) elif decision.action == ACTION_HOLD_ROUTED: hold_routed += 1 rec.reasons.append( "Track C — hold_packet routed to the independent-ANU " "HOLD lane; verify-only (회장 §3.C/§8, reg 8/9/10)." ) else: # ACTION_PROPOSAL_ONLY rec.reasons.append( "unauthorized action -> retained as a proposal, NOT " "enacted (회장 §2.4, reg 3)." ) records.append(rec) if hold: verdict = ENACTOR_HOLD_FOR_CHAIR elif enacted: verdict = ENACTOR_ENACTED elif idem_skip: verdict = ENACTOR_IDEMPOTENT_SKIP else: verdict = ENACTOR_NO_ACTION reasons.append( "bounded enactor: Track A=additive closeout enact only · Track " "B/C=verify-only · merge/PR/credential/branch=blocked · zero " "auto-dispatch/merge/cron/subprocess · idempotent · executor " "self key never owns/operates a callback (회장 §2/§6/§8/§10)." ) return EnactorResult( schema=RESULT_SCHEMA, generated_at_kst=generated_at_kst, verdict=verdict, progress_trigger=TRIGGER_REGISTRY_COMPLETED, fixed_time_used=False, dead_man_used=False, fallback_used=False, authority=AUTHORITY_NONE, loop_result_sha256=loop_sha, proposals_total=len(candidates), enacted_count=enacted, idempotent_skip_count=idem_skip, dispatch_ready_count=dispatch_ready, blocked_count=blocked, hold_routed_count=hold_routed, records=[r.to_json() for r in records], additive_artifacts=additive_artifacts, hold_for_chair=hold, reasons=reasons, ) def enactor_result_envelope( result: EnactorResult, *, generated_at_kst: str = "" ) -> Dict[str, object]: """Serialize the enactor result as the standalone enactor-result artifact (회장 §4 expected_files).""" return { "schema": "task-2553+55.enactor-result.v1", "generated_at_kst": generated_at_kst or result.generated_at_kst, "authority": AUTHORITY_NONE, "action_mode": "bounded_enact", "auto_executed": False, "result": result.to_json(), } __all__ = [ "ENACTOR_SCHEMA", "RESULT_SCHEMA", "CLOSEOUT_ARTIFACT_SCHEMA", "TRIGGER_REGISTRY_COMPLETED", "TRIGGER_FIXED_TIME", "TRIGGER_DEAD_MAN", "TRIGGER_FALLBACK", "FORBIDDEN_PROGRESS_TRIGGERS", "CLOSEOUT_ARTIFACT_TARGET", "AUTHORITY_NONE", "ENACTOR_ENACTED", "ENACTOR_IDEMPOTENT_SKIP", "ENACTOR_NO_ACTION", "ENACTOR_FORBIDDEN_TRIGGER", "ENACTOR_HOLD_FOR_CHAIR", "EnactionRecord", "EnactorResult", "RuntimeEventEnactor", "enactor_result_envelope", ]