# -*- coding: utf-8 -*- """anu_v3.callback_event_trigger — normal-callback durable-success registry write-back + read-only event-driven next_action resolver. task-2553+47 (NORMAL_CALLBACK_REGISTRY_WRITEBACK_AND_EVENT_TRIGGER, 회장 결정). Problem (회장 §2 / +45 실증): the +45 normal-callback collector finished at 14:24 but the durable 4-tuple ledger stayed ``role=dispatch / REGISTERED`` (``normal_collector_cron_id=null``), so ANU could not see the normal callback as completed and the next step / summary was only recovered manually at 14:44. There was no write-back of the verified normal durable-success into ``callback_4tuple_index`` and no read-only event-trigger that detects a COMPLETED record and computes the next action *without* a fixed-time gate or a dead-man fallback as the progress trigger. This module supplies the two missing seams (회장 §3): 1. ``write_back_completed`` — the정정 표준 (a) normal-callback collector session, *after* verification, calls this exactly once to append a correctly-bound ``state=COMPLETED`` ledger record carrying the ``normal_collector_cron_id / fallback_cron_id / task_id / dispatch_id / chat_id / role`` binding. It uses the +44 registry's existing append-only API (``registry.append``) — the +44 module stays byte-0 (additive-only carve-out unused; no escalation of the registry to a merge/dispatch executor, §6). 2. ``CallbackEventTrigger.scan`` — a *read-only* resolver that detects a registry COMPLETED event and, when the dependency is satisfied, computes the next_action. **It never uses a fixed-time gate or a dead-man fallback as the progress trigger** (Layer A, 9-R.1): the only accepted progress trigger is the registry COMPLETED event. A dead-man signal is a missed-callback safety net, not a primary trigger; a fixed-time gate offered as the dependency trigger is rejected with ``FORBIDDEN_TRIGGER_SOURCE`` (regression §8.5). Layer A / NO-CRON (9-R.1): this module performs ZERO cron register/remove, ZERO dispatch, ZERO merge, ZERO ``cokacdir`` / ``subprocess`` exec. The only write surface is the +44 append-only ledger (its designed Layer A role). A next_action for which this session has no write authority is emitted as a **proposal JSON only** — never auto-dispatched / merged / closed out (§6). """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Sequence from anu_v3.callback_4tuple_registry import ( # pyright: ignore[reportMissingImports] # noqa: E501 Callback4TupleRecord, Callback4TupleRegistry, NORMAL_CALLBACK_COMPLETED, NO_LEDGER_RECORD as REG_NO_LEDGER_RECORD, TRACK_MISMATCH as REG_TRACK_MISMATCH, make_record, ) WRITEBACK_SCHEMA = "anu_v3.callback_4tuple_writeback.v1" EVENT_TRIGGER_SCHEMA = "anu_v3.callback_event_trigger.v1" # ── trigger sources ────────────────────────────────────────────────────────── # The ONLY permitted progress / next_action trigger (회장 §3, 9-R.1 Layer A). TRIGGER_REGISTRY_COMPLETED = "registry_completed_event" # Forbidden as a progress / dependency-stage trigger (§6 verbatim). A # fixed-time cron must NEVER drive the next dependent stage. TRIGGER_FIXED_TIME = "fixed_time_gate" # Safety-net ONLY (missed-callback). NOT a primary / progress trigger (§6). TRIGGER_DEAD_MAN = "dead_man_fallback" FORBIDDEN_PROGRESS_TRIGGERS = frozenset( {TRIGGER_FIXED_TIME, TRIGGER_DEAD_MAN} ) # ── write-back classifications ─────────────────────────────────────────────── WRITEBACK_COMPLETED = "WRITEBACK_COMPLETED" WRITEBACK_IDEMPOTENT_SKIP = "WRITEBACK_IDEMPOTENT_SKIP" # normal_collector_cron_id is the MANDATORY executor normal-completion # lifecycle signal (callback mandatory rule, §8.10). Missing -> no write-back, # the rule is NOT weakened here. CALLBACK_MANDATORY_VIOLATION = "CALLBACK_MANDATORY_VIOLATION" # ── next_action verdicts ───────────────────────────────────────────────────── NEXT_ACTION_READY = "NEXT_ACTION_READY" # No COMPLETED event yet — wait. This is a fail-safe DEFER, NOT a progress # trigger and NOT a missing/blocked verdict (registry NO_LEDGER_RECORD). NEXT_ACTION_DEFERRED = "NEXT_ACTION_DEFERRED" # Dependency declared but its registry record is not COMPLETED. NEXT_ACTION_BLOCKED = "NEXT_ACTION_BLOCKED" TRACK_MISMATCH = "TRACK_MISMATCH" # A fixed-time / dead-man source was offered as the dependency trigger. FORBIDDEN_TRIGGER_SOURCE = "FORBIDDEN_TRIGGER_SOURCE" NO_LEDGER_RECORD = "NO_LEDGER_RECORD" # next_action authority: this resolver has NO write/dispatch authority, so # every emitted next_action is a proposal only (회장 §3, §6). AUTHORITY_NONE = "none" ACTION_MODE_PROPOSAL = "proposal" @dataclass(frozen=True) class WritebackResult: """Outcome of a single normal durable-success registry write-back.""" schema: str status: str # WRITEBACK_COMPLETED | WRITEBACK_IDEMPOTENT_SKIP | CALLBACK_MANDATORY_VIOLATION # noqa: E501 task_id: str appended: bool record: Optional[Dict[str, object]] reasons: List[str] = field(default_factory=list) def to_json(self) -> Dict[str, object]: return { "schema": self.schema, "status": self.status, "task_id": self.task_id, "appended": self.appended, "record": self.record, "reasons": list(self.reasons), } def _completed_identity_present( registry: Callback4TupleRegistry, *, task_id: str, dispatch_id: str, chat_id: str, normal_collector_cron_id: str, ) -> bool: """True iff an append-only COMPLETED line with the SAME normal-collector binding already exists (idempotency key — repeated write-back is a skip, never a duplicate append; regression §8.6).""" for r in registry.history_for(task_id): if ( r.status == "COMPLETED" and r.dispatch_id == dispatch_id and r.chat_id == str(chat_id) and r.normal_collector_cron_id == normal_collector_cron_id ): return True return False def write_back_completed( registry: Callback4TupleRegistry, *, task_id: str, dispatch_id: str, dispatch_cron_id: str, executor: str, chat_id: str, normal_collector_cron_id: Optional[str], fallback_callback_cron_id: Optional[str], role: str = "executor", no_fallback: bool = False, ts_kst: str = "", ) -> WritebackResult: """Append a correctly-bound ``state=COMPLETED`` ledger record (회장 §3/§4.1). Called ONCE by the 정정 표준 (a) normal-callback collector session after it has verified the normal durable-success. Unlike a naive ``registry.mark_completed`` (which would copy a stale ``role=dispatch / normal_collector_cron_id=null`` line — the exact +45 14:24→14:44 defect), this binds the *verified* normal-collector identity: ``normal_collector_cron_id / fallback_cron_id / task_id / dispatch_id / chat_id / role``. Append-only & idempotent (§4.4 / §8.6): a repeated call with the same normal-collector binding is a ``WRITEBACK_IDEMPOTENT_SKIP`` — no duplicate line. ``normal_collector_cron_id`` missing -> ``CALLBACK_MANDATORY_ VIOLATION`` with NO append (callback mandatory rule preserved, §8.10). Uses only the +44 registry's existing append-only API — the +44 module stays byte-0; the registry is NOT escalated to a merge/dispatch executor (§6). """ reasons: List[str] = [] if not normal_collector_cron_id: reasons.append( "normal_collector_cron_id missing — MANDATORY executor normal " "completion callback lifecycle signal; write-back refused, " "callback mandatory rule NOT weakened (§8.10)." ) return WritebackResult( schema=WRITEBACK_SCHEMA, status=CALLBACK_MANDATORY_VIOLATION, task_id=task_id, appended=False, record=None, reasons=reasons, ) if _completed_identity_present( registry, task_id=task_id, dispatch_id=dispatch_id, chat_id=chat_id, normal_collector_cron_id=normal_collector_cron_id, ): return WritebackResult( schema=WRITEBACK_SCHEMA, status=WRITEBACK_IDEMPOTENT_SKIP, task_id=task_id, appended=False, record=None, reasons=["COMPLETED record with identical normal-collector " "binding already present — idempotent skip (§8.6)."], ) rec: Callback4TupleRecord = make_record( task_id=task_id, dispatch_id=dispatch_id, dispatch_cron_id=dispatch_cron_id, executor=executor, chat_id=str(chat_id), normal_collector_cron_id=normal_collector_cron_id, fallback_callback_cron_id=fallback_callback_cron_id, role=role, status="COMPLETED", no_fallback=no_fallback, ts_kst=ts_kst, ) registry.append(rec) return WritebackResult( schema=WRITEBACK_SCHEMA, status=WRITEBACK_COMPLETED, task_id=task_id, appended=True, record=rec.to_json(), reasons=[], ) @dataclass(frozen=True) class NextActionResult: """Read-only event-trigger verdict for one task. ``trigger_source`` is ALWAYS the registry COMPLETED event when ``ready`` is True (9-R.1 Layer A). ``fixed_time_used`` / ``dead_man_used`` are explicit FALSE evidence (회장 §9-4 보고 항목). """ schema: str task_id: str verdict: str ready: bool trigger_source: Optional[str] fixed_time_used: bool dead_man_used: bool fallback_pending_non_blocking: bool summary_candidate: bool summary_inputs_completed: List[str] next_action: Optional[Dict[str, object]] reasons: List[str] = field(default_factory=list) def to_json(self) -> Dict[str, object]: return { "schema": self.schema, "task_id": self.task_id, "verdict": self.verdict, "ready": self.ready, "trigger_source": self.trigger_source, "fixed_time_used": self.fixed_time_used, "dead_man_used": self.dead_man_used, "fallback_pending_non_blocking": ( self.fallback_pending_non_blocking ), "summary_candidate": self.summary_candidate, "summary_inputs_completed": list(self.summary_inputs_completed), "next_action": self.next_action, "reasons": list(self.reasons), } class CallbackEventTrigger: """Read-only event-driven next_action resolver over the +44 ledger. ZERO write (the only write surface is ``write_back_completed`` above, explicitly invoked once by the collector). ZERO cron / dispatch / merge / subprocess. ``scan`` is idempotent: repeated scans of the same ledger state yield an identical verdict and never append a record (§8.6). """ def __init__(self, registry: Callback4TupleRegistry) -> None: self.registry = registry # -- read-only event detection ------------------------------------ def _completed_record( self, task_id: str ) -> Optional[Callback4TupleRecord]: rec = self.registry.latest_for(task_id) if rec is not None and rec.status == "COMPLETED": return rec return None def _fallback_pending(self, task_id: str) -> bool: """A fallback cron is bound and still pending (not yet cancelled). Layer A NO-CRON never cancels a fallback, so a bound fallback stays pending after normal durable-success until +45/+48 enacts the verified single cancel. This is surfaced ONLY to ASSERT non-blocking (§8.3): a pending fallback never blocks a completed task's progress. An ABSENT fallback (or an explicit no_fallback contract) returns False — it is never treated as a cancel target (§8.7).""" rec = self.registry.latest_for(task_id) if rec is None: return False if rec.no_fallback or not rec.fallback_callback_cron_id: return False # absent/declared-no fallback -> NOT a cancel target return True def scan( self, *, task_id: str, expected_dispatch_id: Optional[str] = None, expected_chat_id: Optional[str] = None, expected_dispatch_cron_id: Optional[str] = None, dependency_trigger_source: str = TRIGGER_REGISTRY_COMPLETED, consolidated_inputs: Optional[Sequence[str]] = None, next_action_kind: str = "dispatch_next_task", next_action_payload: Optional[Dict[str, object]] = None, dead_man_signal: bool = False, ) -> NextActionResult: """Detect a registry COMPLETED event and compute the next_action. * ``dependency_trigger_source`` MUST be ``registry_completed_event``. A ``fixed_time_gate`` / ``dead_man_fallback`` offered as the dependency trigger -> ``FORBIDDEN_TRIGGER_SOURCE`` (FAIL, §8.5). * No COMPLETED record -> ``NEXT_ACTION_DEFERRED`` (fail-safe wait; a ``dead_man_signal`` alone does NOT promote this to ready — §8.4: dead-man is a safety net, not a primary trigger). * COMPLETED + identity match -> ``NEXT_ACTION_READY``. A pending fallback is recorded non-blocking (§8.3) and an absent fallback is never treated as a cancel target (§8.7). * ``consolidated_inputs`` all COMPLETED -> ``summary_candidate`` emitted immediately (§3 consolidated summary). * The emitted ``next_action`` is a proposal only (no write/dispatch authority — §6); idempotent across repeated scans (§8.6). """ reasons: List[str] = [] # §8.5 — a fixed-time / dead-man source as the dependency-stage # trigger is a hard FAIL (회장 §6 verbatim). if dependency_trigger_source in FORBIDDEN_PROGRESS_TRIGGERS: reasons.append( f"dependency_trigger_source={dependency_trigger_source!r} is " "forbidden as a progress/dependency-stage trigger; only " "registry_completed_event is permitted (회장 §6 / 9-R.1 " "Layer A)." ) return NextActionResult( schema=EVENT_TRIGGER_SCHEMA, task_id=task_id, verdict=FORBIDDEN_TRIGGER_SOURCE, ready=False, trigger_source=None, fixed_time_used=False, dead_man_used=False, fallback_pending_non_blocking=False, summary_candidate=False, summary_inputs_completed=[], next_action=None, reasons=reasons, ) # Registry-first identity / state classification (read-only, +44). verdict_reg = self.registry.classify( task_id=task_id, expected_dispatch_id=expected_dispatch_id, expected_chat_id=expected_chat_id, expected_dispatch_cron_id=expected_dispatch_cron_id, ) if verdict_reg == REG_TRACK_MISMATCH: reasons.append( "registry identity mismatch — an unrelated track's callback " "is never cited as this task's progress (§8.9)." ) return NextActionResult( schema=EVENT_TRIGGER_SCHEMA, task_id=task_id, verdict=TRACK_MISMATCH, ready=False, trigger_source=None, fixed_time_used=False, dead_man_used=False, fallback_pending_non_blocking=False, summary_candidate=False, summary_inputs_completed=[], next_action=None, reasons=reasons, ) completed = self._completed_record(task_id) if verdict_reg != NORMAL_CALLBACK_COMPLETED or completed is None: # No COMPLETED event. dead-man alone must NOT be a primary # trigger (§8.4) — stay deferred (fail-safe wait). if verdict_reg == REG_NO_LEDGER_RECORD: verdict = NO_LEDGER_RECORD reasons.append( "no ledger record — fail-safe DEFER (defer to " "schedule_history + canonical artifact; not a " "progress trigger)." ) else: verdict = NEXT_ACTION_DEFERRED reasons.append( "registry record present but not COMPLETED — wait for " "the normal-callback durable-success write-back (§8.4)." ) if dead_man_signal: reasons.append( "a dead-man signal was observed but is EXPLICITLY NOT " "promoted to a progress trigger — safety net only " "(§8.4, 회장 §6 verbatim). ready stays False." ) return NextActionResult( schema=EVENT_TRIGGER_SCHEMA, task_id=task_id, verdict=verdict, ready=False, trigger_source=None, fixed_time_used=False, # Surface that a dead-man signal was observed but explicitly # NOT promoted to a progress trigger. dead_man_used=False, fallback_pending_non_blocking=False, summary_candidate=False, summary_inputs_completed=[], next_action=None, reasons=reasons, ) # ── COMPLETED event detected — the ONLY permitted progress trigger. fb_pending = self._fallback_pending(task_id) if fb_pending: reasons.append( "fallback binding pending — recorded NON-BLOCKING; the " "completed task proceeds (§8.3). A pending/absent fallback " "is never a cancel target here (§8.7, Layer A NO-CRON)." ) # Consolidated summary: all declared inputs COMPLETED -> emit a # summary candidate immediately (회장 §3). summary_inputs_completed: List[str] = [] summary_candidate = False if consolidated_inputs: for dep in consolidated_inputs: if ( self.registry.classify(task_id=dep) == NORMAL_CALLBACK_COMPLETED ): summary_inputs_completed.append(dep) summary_candidate = len(summary_inputs_completed) == len( list(consolidated_inputs) ) if summary_candidate: reasons.append( "all consolidated inputs COMPLETED — summary candidate " "emitted immediately (회장 §3)." ) next_action = self._build_proposal( task_id=task_id, completed=completed, kind=next_action_kind, payload=next_action_payload, summary_candidate=summary_candidate, summary_inputs_completed=summary_inputs_completed, ) reasons.append( "registry COMPLETED event detected — next_action computed via " "the registry_completed_event trigger ONLY (no fixed-time gate, " "no dead-man fallback; 9-R.1 Layer A)." ) return NextActionResult( schema=EVENT_TRIGGER_SCHEMA, task_id=task_id, verdict=NEXT_ACTION_READY, ready=True, trigger_source=TRIGGER_REGISTRY_COMPLETED, fixed_time_used=False, dead_man_used=False, fallback_pending_non_blocking=fb_pending, summary_candidate=summary_candidate, summary_inputs_completed=summary_inputs_completed, next_action=next_action, reasons=reasons, ) def _build_proposal( self, *, task_id: str, completed: Callback4TupleRecord, kind: str, payload: Optional[Dict[str, object]], summary_candidate: bool, summary_inputs_completed: List[str], ) -> Dict[str, object]: """Build a write-authority-free next_action **proposal** (§6). This resolver never auto-dispatches / merges / closes out. The proposal is data only; an authorized session must enact it. """ return { "schema": "anu_v3.callback_event_trigger.next_action.v1", "authority": AUTHORITY_NONE, "action_mode": ACTION_MODE_PROPOSAL, "auto_executed": False, "kind": kind, "source_task_id": task_id, "trigger_source": TRIGGER_REGISTRY_COMPLETED, "binding": { "task_id": completed.task_id, "dispatch_id": completed.dispatch_id, "dispatch_cron_id": completed.dispatch_cron_id, "normal_collector_cron_id": ( completed.normal_collector_cron_id ), "fallback_callback_cron_id": ( completed.fallback_callback_cron_id ), "chat_id": completed.chat_id, "role": completed.role, }, "summary_candidate": summary_candidate, "summary_inputs_completed": list(summary_inputs_completed), "payload": dict(payload or {}), "note": ( "PROPOSAL ONLY — write 권한 없는 next_action 은 proposal JSON " "으로만 산출(자동 dispatch/merge/closeout 0, 회장 §3/§6). " "An authorized session must enact this." ), } def proposal_envelope( result: NextActionResult, *, generated_at_kst: str = "" ) -> Dict[str, object]: """Serialize a NextActionResult as the standalone next-action-proposal artifact (회장 §5 expected_files).""" return { "schema": "task-2553+47.next-action-proposal.v1", "generated_at_kst": generated_at_kst, "authority": AUTHORITY_NONE, "action_mode": ACTION_MODE_PROPOSAL, "auto_executed": False, "result": result.to_json(), } def load_registry(ledger_path) -> Callback4TupleRegistry: """Convenience: build a read-only registry over an explicit ledger path.""" return Callback4TupleRegistry(Path(ledger_path)) __all__ = [ "WRITEBACK_SCHEMA", "EVENT_TRIGGER_SCHEMA", "TRIGGER_REGISTRY_COMPLETED", "TRIGGER_FIXED_TIME", "TRIGGER_DEAD_MAN", "FORBIDDEN_PROGRESS_TRIGGERS", "WRITEBACK_COMPLETED", "WRITEBACK_IDEMPOTENT_SKIP", "CALLBACK_MANDATORY_VIOLATION", "NEXT_ACTION_READY", "NEXT_ACTION_DEFERRED", "NEXT_ACTION_BLOCKED", "TRACK_MISMATCH", "FORBIDDEN_TRIGGER_SOURCE", "NO_LEDGER_RECORD", "WritebackResult", "NextActionResult", "write_back_completed", "CallbackEventTrigger", "proposal_envelope", "load_registry", ]