# -*- coding: utf-8 -*- """dispatch.finalize_hooks — executor completion lifecycle 결선. task-2635 — normal callback registration enforcement (Core hardening). task-2635+1 — build-then-register-then-update 순서 + 5축 propagation. Spec: memory/specs/system_normal_callback_registration_implementation_spec_260523.md sha256: 0fbd1dad1e110c49474dfbdf13a21fb3bdd9c7f094128004dba8472840bb832d 회장 verbatim (task-2635 §3.3): dispatch/finalize_hooks.py — executor 완료 lifecycle 결선 * finalize_with_callback_registration(task_id, result, anu_key) -> finalize_result * 호출 순서: envelope build → registrar 호출 → registration_status 회수 → result.json 갱신 → fail-closed 분기 * sendfile 은 별도 함수 send_envelope_to_chat(envelope, chat_id) — callback 대체 아님 (보조) 회장 verbatim (task-2635+1 §4): register 성공 후 envelope payload 갱신 — registration_result_status = REGISTERED · cron_schedule_id = 실제 id · registered_at_ts = 실제 시각 · attempted_callback_registration = True · delivery_method = anu_cron_callback. ANCHOR-1 (task-2635+1): envelope payload 최종 5축 상태 정확 반영 — build-then-register-then-update 순서. ANCHOR-3: sendfile/report ≠ callback — 함수/스키마 분리. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional from utils.anu_callback_fallback import ( FallbackDecision, decide_fallback_cancel, expected_collector_spawn, ) from utils.anu_callback_registrar import ( INDEPENDENT_ANU_KEY, RegistrarResult, build_callback_envelope, merge_registrar_result_into_envelope, register_normal_callback, ) from utils.callback_envelope_schema import ( CANONICAL_ROOT_DEFAULT, CallbackDeliveryStatus, CollectorReceiptStatus, NormalCallbackRegistrationStatus, RegistrationResultStatus, detect_status_contradictions, is_callback_complete, is_fail_closed_status, is_success_status, validate_envelope, ) FINALIZE_HOOK_SCHEMA = "dispatch.finalize_hooks.v2" # v2 = 5축 분리 (task-2635+1) @dataclass class FinalizeResult: """Lifecycle outcome returned to the executor. ``finalize_result`` ∈ {"success", "fail"}. The executor MUST translate "fail" to a non-zero exit code (fail-closed contract, spec §2). """ finalize_result: str registration_status: str # axis-3 value (legacy field name retained) fallback_decision: FallbackDecision envelope: Dict[str, Any] registrar_argv: List[str] = field(default_factory=list) is_callback_complete: bool = False is_collector_spawn_expected: bool = False errors: List[str] = field(default_factory=list) notes: List[str] = field(default_factory=list) # task-2635+1 — 5-axis snapshot for upstream consumers registration_intent: bool = True registration_attempted: bool = False registration_result_status: str = "" callback_delivery_status: str = "" collector_receipt_status: str = "" def _validate_or_collect_errors(envelope: Dict[str, Any]) -> List[str]: """Return validation errors (empty on PASS). Validator allows the seed NOT_REGISTERED status with no cron_schedule_id — that's intentional for fail-closed paths (envelope-only / sendfile_only callers).""" _, errs = validate_envelope(envelope) return errs def _five_axis_snapshot(envelope: Dict[str, Any]) -> Dict[str, Any]: """Read the 5 axes off an envelope, with safe fallbacks.""" return { "registration_intent": bool(envelope.get("registration_intent", True)), "registration_attempted": bool(envelope.get("registration_attempted", False)), "registration_result_status": envelope.get( "registration_result_status", envelope.get("registration_status", "") ) or "", "callback_delivery_status": envelope.get( "callback_delivery_status", CallbackDeliveryStatus.NOT_APPLICABLE.value ), "collector_receipt_status": envelope.get( "collector_receipt_status", CollectorReceiptStatus.NOT_APPLICABLE.value ), } def finalize_with_callback_registration( task_id: str, result: Dict[str, Any], anu_key: str = INDEPENDENT_ANU_KEY, *, delay_seconds: int = 10, subprocess_runner: Optional[Callable[[List[str], int], Any]] = None, skip_registration: bool = False, skip_reason: Optional[str] = None, delivery_method_override: Optional[str] = None, seed_envelope_overrides: Optional[Dict[str, Any]] = None, cli_exists_check: Optional[Callable[[str], bool]] = None, at_value: Optional[str] = None, canonical_root: Optional[str] = CANONICAL_ROOT_DEFAULT, ) -> FinalizeResult: """task-2635+1 §4 — build-then-register-then-update pipeline. 1. Build seed envelope (5 axes pre-populated for NOT_REGISTERED + PENDING). 2. If skip_registration → SKIPPED_WITH_EXPLICIT_REASON pathway. 3. Else → call registrar → merge result back into envelope (axes 3/4/5 transition together). 4. Run contradiction scan + envelope validator → fail-closed gating. 5. Return FinalizeResult with full 5-axis snapshot. """ if skip_registration: if not skip_reason: # 최소 envelope (Gemini medium 대응: downstream KeyError 방지 — task_id/anu_key/collector_role 등 기본키 유지). minimal_envelope = { "task_id": task_id, "anu_key": anu_key, "collector_role": "ANU", "registration_intent": True, "registration_attempted": False, "registration_result_status": RegistrationResultStatus.NOT_REGISTERED.value, "registration_status": RegistrationResultStatus.NOT_REGISTERED.value, "callback_delivery_status": CallbackDeliveryStatus.NOT_APPLICABLE.value, "collector_receipt_status": CollectorReceiptStatus.NOT_APPLICABLE.value, "attempted_callback_registration": False, "delivery_method": delivery_method_override or "none", } if canonical_root: minimal_envelope["canonical_root"] = canonical_root return FinalizeResult( finalize_result="fail", registration_status=RegistrationResultStatus.NOT_REGISTERED.value, fallback_decision=FallbackDecision( cancel_fallback=False, reason="skip_registration requested without skip_reason", durable_success_marker_present=False, status=RegistrationResultStatus.NOT_REGISTERED.value, ), envelope=minimal_envelope, errors=["skip_registration=True requires a non-empty skip_reason"], registration_intent=True, registration_attempted=False, registration_result_status=RegistrationResultStatus.NOT_REGISTERED.value, callback_delivery_status=CallbackDeliveryStatus.NOT_APPLICABLE.value, collector_receipt_status=CollectorReceiptStatus.NOT_APPLICABLE.value, ) envelope = build_callback_envelope( task_id=task_id, result=result, anu_key=anu_key, attempted_callback_registration=False, delivery_method=delivery_method_override or "none", registration_status=RegistrationResultStatus.SKIPPED_WITH_EXPLICIT_REASON.value, explicit_skip_reason=skip_reason, registration_intent=True, registration_attempted=False, ) if canonical_root: envelope["canonical_root"] = canonical_root if seed_envelope_overrides: envelope.update(seed_envelope_overrides) errors = _validate_or_collect_errors(envelope) decision = decide_fallback_cancel(envelope) status = envelope["registration_result_status"] snap = _five_axis_snapshot(envelope) return FinalizeResult( finalize_result="success" if is_success_status(status) and not errors else "fail", registration_status=status, fallback_decision=decision, envelope=envelope, is_callback_complete=is_callback_complete(envelope), is_collector_spawn_expected=expected_collector_spawn(envelope), errors=errors, notes=["skipped registration with explicit reason"], **snap, ) # task-2635+1 §4: build → register → update (in this strict order). seed_envelope = build_callback_envelope( task_id=task_id, result=result, anu_key=anu_key, delivery_method=delivery_method_override or "anu_cron_callback", attempted_callback_registration=True, registration_intent=True, registration_attempted=True, ) if canonical_root: seed_envelope["canonical_root"] = canonical_root if seed_envelope_overrides: seed_envelope.update(seed_envelope_overrides) registrar_result: RegistrarResult = register_normal_callback( envelope=seed_envelope, delay_seconds=delay_seconds, anu_key=anu_key, subprocess_runner=subprocess_runner, cli_exists_check=cli_exists_check, at_value=at_value, ) # ANCHOR-1 — register-after-update: only NOW do we mutate the envelope's # status axes to reflect the registrar outcome (REGISTERED/schedule_id # only when the CLI actually returned ok). final_envelope = merge_registrar_result_into_envelope(seed_envelope, registrar_result) errors = _validate_or_collect_errors(final_envelope) contradictions = detect_status_contradictions(final_envelope) decision = decide_fallback_cancel(final_envelope) status = final_envelope["registration_result_status"] if is_fail_closed_status(status): outcome = "fail" elif errors or contradictions: outcome = "fail" else: outcome = "success" snap = _five_axis_snapshot(final_envelope) return FinalizeResult( finalize_result=outcome, registration_status=status, fallback_decision=decision, envelope=final_envelope, registrar_argv=registrar_result.argv, is_callback_complete=is_callback_complete(final_envelope), is_collector_spawn_expected=expected_collector_spawn(final_envelope), errors=errors + contradictions, notes=[ f"registrar_byte_count={registrar_result.byte_count}", f"registrar_error={registrar_result.error or 'none'}", ], **snap, ) def send_envelope_to_chat( envelope: Dict[str, Any], chat_id: str, sendfile_runner: Optional[Callable[[Dict[str, Any], str], Any]] = None, ) -> Dict[str, Any]: """Sendfile-only auxiliary delivery (ANCHOR-3 separation). THIS IS NOT A SUBSTITUTE FOR ``register_normal_callback``. It exists so operators have a structured place to call envelope sendfile, but it does NOT register a cron, does NOT spawn a collector, and does NOT mark ``registration_result_status`` as REGISTERED. The function returns an audit record only; the registrar remains the single source of truth for callback registration. """ def _noop_sendfile(env: Dict[str, Any], cid: str) -> Dict[str, Any]: return {"status": "noop", "envelope_bytes": len(str(env)), "chat_id": cid} runner = sendfile_runner if sendfile_runner is not None else _noop_sendfile outcome = runner(envelope, chat_id) return { "schema": FINALIZE_HOOK_SCHEMA + ".sendfile", "delivered_to": chat_id, "outcome": outcome, "is_callback_substitute": False, "note": ( "sendfile is auxiliary (ANCHOR-3). register_normal_callback " "is the only path that produces a REGISTERED status." ), }