# -*- coding: utf-8 -*- """anu_v3.checkpoint_turn_boundary_sweep — PURE READ-ONLY turn-boundary sweep. task-2553+43 STEP 3 (회장 GO) — runtime checkpoint operationalization 후보. OPERATIONALIZATION CANDIDATE ONLY. This module is the read-only detection *candidate* that ANU *could* call at a turn boundary so a finished +31 checkpoint state (drift / stale / recovery / NO-CRON done / result-ready with no normal callback) is enumerated WITHOUT a chair question. It does NOT wire itself anywhere; actual operational wiring is a separate post-task chair GO (task §5 "실 운영 결선 적용" forbidden). DESIGN BOUNDARY — 9-R.1 Layer A (task §1/§3/§4/§5; regression static+dynamic): * PURE read-only function — checkpoint state read -> candidate enumeration * ZERO write / cron register|remove / merge / PR / dispatch / closeout * idempotent — N calls with identical inputs return identical output * +31 runtime_reconcile_checkpoint.py + recovery layer consumed READ-ONLY (import + emit=False entrypoint only; emit=True is NEVER reached) * recovery-not-primary invariant asserted via the +32 sidecar (read-only): callback primary / fallback safety / cancel-on-success paths preserved, never replaced (task §5 "fallback safety path 제거" / "primary callback 대체물로 격상" forbidden) * 9-R.1 Layer B (executor §8 completion callback cron) is a SEPARATE process-lifecycle signal fired by the executor via external cron tooling — NOT this module's behaviour and NOT this module's side-effect. This module never registers/removes a cron, never writes a file, never merges, never opens a PR, never mutates the +31 frozen original. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Dict, List from anu_v3.runtime_reconcile_checkpoint import RuntimeReconcileCheckpoint from anu_v3.runtime_reconcile_checkpoint import ( DEFAULT_STALE_SECONDS as _CKPT_DEFAULT_STALE_SECONDS, ) from anu_v3.runtime_next_action_resolver import ( NO_CRON_TASK_DONE, RESULT_READY_NO_NORMAL_CALLBACK, NORMAL_COLLECTOR_COMPLETED, DUPLICATE_CALLBACK_IGNORED, TRACK_MISMATCH, RUNNING, FALLBACK_PENDING, STALE_OR_BOT_STUCK_CANDIDATE, ) from anu_v3 import runtime_reconcile_checkpoint_recovery_layer as _recovery SWEEP_RESULT_SCHEMA = "anu_v3.checkpoint_turn_boundary_sweep.result.v1" WIRING_CANDIDATE_SCHEMA = ( "anu_v3.checkpoint_turn_boundary_sweep.wiring_candidate.v1" ) # Map of +31 taxonomy classification -> sweep candidate bucket. Terminal # states that need an ANU follow-up are "actionable candidates"; benign / # in-flight states are enumerated but produce zero action recommendation. _CANDIDATE_TYPE = { NO_CRON_TASK_DONE: "NO_CRON_DONE_CANDIDATE", RESULT_READY_NO_NORMAL_CALLBACK: "RECOVERY_CANDIDATE", STALE_OR_BOT_STUCK_CANDIDATE: "STALE_CANDIDATE", TRACK_MISMATCH: "DRIFT_CANDIDATE", NORMAL_COLLECTOR_COMPLETED: "BENIGN_NORMAL_COMPLETED", DUPLICATE_CALLBACK_IGNORED: "BENIGN_DUPLICATE_IGNORED", FALLBACK_PENDING: "INFLIGHT_FALLBACK_PENDING", RUNNING: "INFLIGHT_RUNNING", } _ACTIONABLE = frozenset({ "NO_CRON_DONE_CANDIDATE", "RECOVERY_CANDIDATE", "STALE_CANDIDATE", "DRIFT_CANDIDATE", }) @dataclass(frozen=True) class SweepCandidate: """One read-only enumerated turn-boundary sweep candidate.""" task_id: str classification: str candidate_type: str actionable: bool recovery_eligible: bool terminal: bool recommended_signal: str # recommendation string ONLY — never executed def to_json(self) -> Dict[str, object]: return { "task_id": self.task_id, "classification": self.classification, "candidate_type": self.candidate_type, "actionable": self.actionable, "recovery_eligible": self.recovery_eligible, "terminal": self.terminal, "recommended_signal": self.recommended_signal, } # Recommendation strings are advisory text only. They are NEVER dispatched, # NEVER registered as a cron, NEVER executed by this module. _RECOMMENDED_SIGNAL = { "NO_CRON_DONE_CANDIDATE": "ANU may recognize NO-CRON completion read-only (no chair question)", "RECOVERY_CANDIDATE": "recovery layer eligible — surface for ANU follow-up (advisory only)", "STALE_CANDIDATE": "stale / bot-stuck candidate — surface for ANU review (advisory)", "DRIFT_CANDIDATE": "4-tuple drift candidate — surface for ANU review (advisory)", "BENIGN_NORMAL_COMPLETED": "no action — normal collector completed", "BENIGN_DUPLICATE_IGNORED": "no action — duplicate callback ignored", "INFLIGHT_FALLBACK_PENDING": "no action — fallback pending (in-flight)", "INFLIGHT_RUNNING": "no action — running (in-flight)", } def recovery_invariant_status() -> Dict[str, object]: """Read-only assertion that the +31 checkpoint stays recovery-not-primary. Delegates entirely to the frozen +32 sidecar (imported read-only). Proves the callback primary / fallback safety paths are preserved, never replaced (task §5). No mutation, no I/O. """ violations = _recovery.assert_checkpoint_is_recovery_not_primary() contract = _recovery.recovery_layer_contract() return { "recovery_not_primary_ok": not violations, "violations": list(violations), "replaces_callback_primary_path": _recovery.checkpoint_replaces_callback_primary_path(), "discards_fallback_safety_path": _recovery.checkpoint_discards_fallback_safety_path(), "contract": contract, } def _zero_side_effect_proof() -> Dict[str, object]: return { "write": 0, "cron_register": 0, "cron_remove": 0, "merge": 0, "pr": 0, "dispatch": 0, "closeout_confirm": 0, "callback_collector_path_touched": 0, "frozen_plus31_mutation": 0, "consumes_plus31_via": "import + checkpoint_entrypoint(emit=False)", "emit_true_reached": False, "note": "Layer A (9-R.1): pure read-only enumeration. The §8 " "executor completion callback cron is Layer B — a separate " "process-lifecycle signal fired by the executor via external " "cron tooling, NOT this module's behaviour.", } def sweep_turn_boundary( repo_root: Path, fixture_path: Path, *, generated_ts_kst: str = "", stale_seconds: int = _CKPT_DEFAULT_STALE_SECONDS, ) -> Dict[str, object]: """PURE read-only turn-boundary sweep. Reads +31 checkpoint state for ``fixture_path`` via the read-only ``checkpoint_entrypoint`` (emit=False) and enumerates drift / stale / recovery / NO-CRON-done / result-ready candidates. Returns a fully deterministic document — identical inputs => byte-identical output (idempotent). Writes NOTHING, registers/removes NO cron, merges NOTHING. ``generated_ts_kst`` is passed straight through (caller-supplied, default empty) so the function holds no internal clock — a precondition of the idempotence guarantee. """ repo_root = Path(repo_root) fixture_path = Path(fixture_path) # Read-only consumption of +31. emit=False — emit=True is unreachable # from this module (task §5 "실 운영 결선 / write" forbidden). ckpt = RuntimeReconcileCheckpoint(repo_root, stale_seconds=stale_seconds) packet = ckpt.checkpoint_entrypoint( fixture_path, generated_ts_kst=generated_ts_kst ) decision = packet["decision_packet"] track_records: Dict[str, dict] = decision["track_records"] # type: ignore candidates: List[SweepCandidate] = [] for tid in sorted(track_records): # sorted => deterministic order rec = track_records[tid] classification = str(rec["classification"]) ctype = _CANDIDATE_TYPE.get(classification, "UNKNOWN") candidates.append(SweepCandidate( task_id=tid, classification=classification, candidate_type=ctype, actionable=ctype in _ACTIONABLE, recovery_eligible=bool(rec.get("recovery_eligible", False)), terminal=bool(rec.get("terminal", False)), recommended_signal=_RECOMMENDED_SIGNAL.get(ctype, ""), )) by_type: Dict[str, List[str]] = {} for c in candidates: by_type.setdefault(c.candidate_type, []).append(c.task_id) actionable = [c.task_id for c in candidates if c.actionable] return { "schema": SWEEP_RESULT_SCHEMA, "task_id": "task-2553+43", "sweep": "CHECKPOINT_TURN_BOUNDARY_SWEEP", "mode": "READ_ONLY", "layer": "A_DELIVERABLE_PURE_READ_ONLY (9-R.1)", "generated_ts_kst": generated_ts_kst, "source_fixture": str(fixture_path), "consumes_plus31": { "module": "anu_v3.runtime_reconcile_checkpoint (+31, frozen)", "entrypoint": "checkpoint_entrypoint(emit=False)", "recovery_layer": "anu_v3." "runtime_reconcile_checkpoint_recovery_layer (+32, frozen)", }, "candidates": [c.to_json() for c in candidates], "by_candidate_type": by_type, "actionable_candidates": actionable, "actionable_count": len(actionable), "tracks_total": len(candidates), "recovery_invariant": recovery_invariant_status(), "zero_side_effect_proof": _zero_side_effect_proof(), } def wiring_candidates() -> Dict[str, object]: """Machine-readable operational-wiring CANDIDATE enumeration. Pure constant data describing hook points where ``sweep_turn_boundary`` *could* be wired at an ANU turn boundary. Each entry proves read-only / side-effect-0 and records conflict + risk tier. NOTHING is wired here — actual operational wiring is a separate post-task chair GO (task §5 "실 운영 결선 적용" forbidden). Returning this dict has no side effect. """ return { "schema": WIRING_CANDIDATE_SCHEMA, "task_id": "task-2553+43", "applied": False, "note": "CANDIDATES ONLY — zero wiring applied (task §5). Each entry " "is read-only & side-effect-0; actual wiring needs a " "separate chair GO post-task.", "candidates": [ { "id": "WC1", "hook_point": "ANU active-dispatch turn boundary " "(pre-response)", "call_site": "ANU turn loop, before composing the reply: " "call sweep_turn_boundary(repo_root, " "active_dispatch_fixture, emit unreachable)", "read_only_guarantee": "sweep_turn_boundary delegates to +31 " "checkpoint_entrypoint(emit=False); no write/cron/merge path " "exists in the call graph", "side_effect_zero_proof": "regression static scan (forbidden " "write/cron/merge/subprocess tokens absent) + dynamic " "write/cron sentinel (0 captured) + idempotent N-call equality", "conflict": "none — additive read-only call; does not touch " "callback/collector/fallback paths", "risk_tier": "LOW", }, { "id": "WC2", "hook_point": "ANU pre-answer reconcile gate " "(§11 결선, advisory)", "call_site": "ANU, before answering a chair query about " "task state: consult sweep actionable_candidates " "to self-detect NO-CRON completion", "read_only_guarantee": "same read-only call graph as WC1; " "recommended_signal strings are advisory text, never executed", "side_effect_zero_proof": "shared with WC1 regression " "(static+dynamic+idempotent)", "conflict": "none — does not replace the normal completion " "callback primary path (recovery_invariant asserts " "recovery-not-primary, task §5)", "risk_tier": "LOW", }, { "id": "WC3", "hook_point": "external scheduled read-only audit " "(out-of-band, advisory)", "call_site": "an EXISTING external scheduler (NOT registered " "by this module) invokes sweep_turn_boundary " "read-only for an out-of-band drift/stale audit", "read_only_guarantee": "this module neither registers nor " "removes that schedule (task §5 'cron 등록·제거'); it only " "exposes a pure function the external tool may call", "side_effect_zero_proof": "shared with WC1 regression; the " "schedule itself is owned by external tooling (Layer B-like), " "not this Layer-A module", "conflict": "must NOT be conflated with §8 executor " "completion callback (separate lifecycle signal, 9-R.1 " "Layer B)", "risk_tier": "MEDIUM", }, ], "cross_track_isolation": { "parallel_task": "task-2553+42 (Step 2 adoption 준비)", "expected_files_disjoint": True, "other_track_originals_modified": 0, }, }