"""utils/completion_callback_operational_cancel_seam.py task-2553+23 — CALLBACK_CANCEL_ON_SUCCESS_OPERATIONAL_INTEGRATION. 운영 collector 통합 seam: normal completion-callback collector 가 durable evidence 성공 판정을 낸 **직후**, 사전등록 fallback callback cron 을 운영 경로로 자동 취소한다. (normal 성공 → fallback cron remove → 불필요 fallback 발화 0) 결선 순서 (task-2553+23 §4.3): 1. durable-evidence 성공 게이트 (task-2553+9a evaluate_durable_evidence 상속) └ 미충족 → SKIP, fallback 보존, normal 성공 불변 2. **신규** live cron-state verifier (utils/live_cron_state_verifier) └ VERIFIED 아님 → SKIP, 실 remove 0, fallback 보존 3. VERIFIED → task-2553+9a cancel_fallback_on_success 호출 (운영 remover DI = utils.completion_callback_fallback_cancel. RealCokacdirCronRemover, 운영 런타임은 dry_run=False) └ +9a 가 §9-R.1 safe-remove 5조건 + §9-R.2 durable 재확인 후 실 remove 4. audit trail JSON 기록 (모든 remove 결정·5조건·invoked/skip·사유) 5. outcome 반환 — **normal collector 성공/보고는 본 결과와 독립** (remove 실패·skip 이어도 normal 성공 유지, collector 실패 처리 0) 설계 규율: * frozen anchor utils/anu_delegation_completion_callback.py 무접촉 — import/호출 0 (AST 정적 증명 대상). * task-2553+9a 분리 모듈을 **수정 없이 사용**. 운영 remover 는 +9a 가 이미 제공하는 RealCokacdirCronRemover(dry_run=False) 주입점을 재사용한다 (§4.1 "신규 옵션/주입점만, 기존 dry-run 시그니처·동작 무회귀" → +9a 무변경으로 strict-additive 충족, 본 seam 이 운영 결선). * 본 task 구현/테스트는 실 운영 cron 절대 비접촉(§6 9-R.4): 기본 operational=False(dry_run 경로). 운영 collector 만 operational=True. * 디커플: 어떤 분기에서도 예외를 collector 로 전파하지 않는다 — 항상 OperationalCancelOutcome 반환, normal_success_preserved=True. """ from __future__ import annotations import json from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Callable, Optional from utils.completion_callback_fallback_cancel import ( CancelClassification, RealCokacdirCronRemover, Remover, cancel_fallback_on_success, evaluate_durable_evidence, ) from utils.live_cron_state_verifier import ( CronLister, verify_live_cron_state, ) def _now_utc() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") @dataclass class OperationalCancelOutcome: """seam 결과. normal collector 성공은 본 결과와 **독립**.""" stage: str # durable_gate | live_verify | plus9a_cancel seam_classification: str task_id: str target_cron_id: str cron_remove_invoked: bool fallback_cancelled: bool remove_allowed_by_live_verifier: bool normal_success_preserved: bool skip_reason: str durable_evidence: dict = field(default_factory=dict) live_verification: dict = field(default_factory=dict) plus9a_decision: Optional[dict] = None audit_path: Optional[str] = None notes: list = field(default_factory=list) ts_utc: str = "" def to_dict(self) -> dict: return { "schema": "operational_cancel_seam_outcome_v1", "task_id": self.task_id, "target_cron_id": self.target_cron_id, "stage": self.stage, "seam_classification": self.seam_classification, "cron_remove_invoked": self.cron_remove_invoked, "fallback_cancelled": self.fallback_cancelled, "remove_allowed_by_live_verifier": self.remove_allowed_by_live_verifier, "normal_success_preserved": self.normal_success_preserved, "skip_reason": self.skip_reason, "durable_evidence": self.durable_evidence, "live_verification": self.live_verification, "plus9a_decision": self.plus9a_decision, "audit_path": self.audit_path, "notes": self.notes, "ts_utc": self.ts_utc, } def _write_audit(audit_path: Optional[Path], record: dict) -> Optional[str]: if audit_path is None: return None try: audit_path.parent.mkdir(parents=True, exist_ok=True) audit_path.write_text( json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8" ) return str(audit_path) except OSError: return None def run_operational_cancel_seam( *, task_id: str, target_cron_id: str, dispatch_fired_marker_path: Path, result_json_path: Path, report_path: Path, collector_result_marker_path: Path, audit_path: Optional[Path] = None, cron_lister: Optional[CronLister] = None, remover: Optional[Remover] = None, fallback_cancelled_marker_path: Optional[Path] = None, cancel_lock_path: Optional[Path] = None, callback_contract: Optional[dict] = None, operational: bool = False, now_fn: Callable[[], str] = _now_utc, ) -> OperationalCancelOutcome: """운영 collector 통합 seam 진입점. normal collector 성공/보고는 본 함수 결과와 **독립**이다. 본 함수는 어떤 경우에도 예외를 호출자(collector)로 전파하지 않으며 항상 ``normal_success_preserved=True`` 인 outcome 을 반환한다. operational=False(기본) → +9a 에 dry_run=True 전달(실 cron 비접촉, 본 task 구현/테스트 안전 기본값). 운영 collector 만 operational=True(dry_run=False). """ # 디커플 절대불변: 시계(now_fn) 가 예외를 던져도 collector 로 전파 0 — # placeholder ts 로 강등하여 audit/outcome 는 항상 생성한다. def _safe_ts() -> str: try: return now_fn() except Exception: # noqa: BLE001 - now_fn 실패도 비전파(디커플) return "0000-00-00T00:00:00Z" ts = _safe_ts() dry_run = not operational def _finish(outcome: OperationalCancelOutcome) -> OperationalCancelOutcome: outcome.normal_success_preserved = True outcome.ts_utc = ts record = { "schema": "task-2553+23.cancel-audit_v1", "task_id": task_id, "target_cron_id": target_cron_id, "decision": outcome.to_dict(), "operational": operational, "ts_utc": ts, } outcome.audit_path = _write_audit(audit_path, record) return outcome try: # ── 1. durable-evidence 성공 게이트 (+9a 상속) ────────────────────── ev = evaluate_durable_evidence( result_json_path=result_json_path, report_path=report_path, collector_result_marker_path=collector_result_marker_path, ) if not ev["satisfied"]: return _finish( OperationalCancelOutcome( stage="durable_gate", seam_classification="SKIP_DURABLE_NOT_SATISFIED", task_id=task_id, target_cron_id=target_cron_id, cron_remove_invoked=False, fallback_cancelled=False, remove_allowed_by_live_verifier=False, normal_success_preserved=True, skip_reason=( "normal collector durable evidence 미충족 → fallback " f"보존(예정대로 발화): {ev['reason']}" ), durable_evidence=ev, notes=[ "디커플: durable 미충족은 normal 성공과 무관 — collector " "실패 처리 0", ], ) ) # ── 2. 신규 live cron-state verifier ─────────────────────────────── lv = verify_live_cron_state( task_id=task_id, target_cron_id=target_cron_id, dispatch_fired_marker_path=dispatch_fired_marker_path, cron_lister=cron_lister, now_fn=now_fn, ) if not lv.remove_allowed: return _finish( OperationalCancelOutcome( stage="live_verify", seam_classification=f"SKIP_LIVE_{lv.classification.value}", task_id=task_id, target_cron_id=target_cron_id, cron_remove_invoked=False, fallback_cancelled=False, remove_allowed_by_live_verifier=False, normal_success_preserved=True, skip_reason=lv.skip_reason, durable_evidence=ev, live_verification=lv.to_dict(), notes=[ "live 조회 권위 — 5조건 미충족/조회실패/idempotent 시 실 " "remove 0, fallback 보존", ], ) ) # ── 3. VERIFIED → +9a cancel_fallback_on_success (운영 remover DI) ── active_remover: Remover = remover or RealCokacdirCronRemover() decision = cancel_fallback_on_success( task_id=task_id, target_cron_id=target_cron_id, dispatch_fired_marker_path=dispatch_fired_marker_path, result_json_path=result_json_path, report_path=report_path, collector_result_marker_path=collector_result_marker_path, fallback_cancelled_marker_path=fallback_cancelled_marker_path, cancel_lock_path=cancel_lock_path, callback_contract=callback_contract, normal_collector_success=True, remover=active_remover, dry_run=dry_run, now_fn=now_fn, ) cancelled = decision.classification == CancelClassification.CANCELLED return _finish( OperationalCancelOutcome( stage="plus9a_cancel", seam_classification=f"PLUS9A_{decision.classification.value}", task_id=task_id, target_cron_id=target_cron_id, cron_remove_invoked=decision.cron_remove_invoked, fallback_cancelled=decision.fallback_cancelled, remove_allowed_by_live_verifier=True, normal_success_preserved=True, skip_reason=decision.cancel_skipped_reason, durable_evidence=ev, live_verification=lv.to_dict(), plus9a_decision=decision.to_dict(), notes=[ "live verifier VERIFIED 후 +9a 가 §9-R.1/§9-R.2 재확인 후 " "실 remove (운영 remover DI)", ( "fallback cron 자동취소 성공 → 불필요 fallback 발화 0" if cancelled else "remove 비성공이어도 normal 성공 불변 (디커플)" ), ], ) ) except Exception as exc: # noqa: BLE001 - 디커플: 어떤 예외도 collector 비전파 return _finish( OperationalCancelOutcome( stage="exception_guard", seam_classification="SKIP_SEAM_EXCEPTION", task_id=task_id, target_cron_id=target_cron_id, cron_remove_invoked=False, fallback_cancelled=False, remove_allowed_by_live_verifier=False, normal_success_preserved=True, skip_reason=f"seam 내부 예외 → fallback 보존, normal 성공 불변: {exc}", notes=["디커플: seam 예외는 collector 실패로 전파 0"], ) )