"""anu_v2.polling_policy — long polling 금지 코드 게이트 (task-2556 §9 / §12). 회장 §명시 2026-05-12 KST (task-2556 12 필수 §9): "long polling 금지 — 5~15min normal wait / 30min first timeout / 1회 recheck / 봇 종료" 본 모듈은 ``ExecutorScheduler`` / ``IdlePRDiagnoser`` 가 의도치 않게 long polling 루프로 변질되는 것을 **코드 레벨에서 정적/동적 게이트** 로 차단한다. 설계 원칙: - ``MAX_SINGLE_SLEEP_SECONDS = 900`` (15 분) — 단일 호출의 sleep 상한. - ``FIRST_TIMEOUT_SECONDS = 1800`` (30 분) — 첫 trigger 후 fresh 미도착 대기 상한. - ``MAX_RECHECKS = 1`` — fresh 도착 여부 재확인 1 회만. - ``BotSessionExitRequired`` sentinel — recheck 1 회 후 봇은 즉시 종료해야 함을 알린다. 부수효과 0: 본 모듈은 sleep 자체를 수행하지 않고, **정책 위반 시 예외만 raise** 한다. 호출자는 본 모듈의 ``assert_sleep_allowed`` / ``advance_recheck`` 만 사용하면 long polling 패턴 (while True + 짧은 sleep 반복) 을 코드 레벨에서 표현 불가. one-way isolation: anu_v2/ 외부 import 금지. stdlib + ``dataclasses`` 만 사용. """ from __future__ import annotations from dataclasses import dataclass, replace from typing import Any, Final MAX_SINGLE_SLEEP_SECONDS: Final[int] = 900 # 15 minutes FIRST_TIMEOUT_SECONDS: Final[int] = 1800 # 30 minutes MAX_RECHECKS: Final[int] = 1 NORMAL_WAIT_MIN_SECONDS: Final[int] = 300 # 5 minutes NORMAL_WAIT_MAX_SECONDS: Final[int] = 900 # 15 minutes class LongPollingViolation(RuntimeError): """long polling pattern detected — task-2556 §9 hard-block.""" class BotSessionExitRequired(RuntimeError): """recheck 1 회 후 봇은 즉시 종료해야 함. 호출자는 본 예외를 catch 후 process exit.""" @dataclass(frozen=True) class PollingState: """recheck 상태 머신 (immutable, replace 로 다음 state 생성). Attributes: rechecks_done: 지금까지 수행된 recheck 횟수. elapsed_seconds: trigger 후 누적 경과 시간 (외부에서 측정 후 주입). posted_marker_present: owner-trigger.posted marker 존재 여부. """ rechecks_done: int = 0 elapsed_seconds: int = 0 posted_marker_present: bool = False def assert_sleep_allowed(seconds: int) -> None: """단일 sleep 호출 상한 검증. ``seconds > 900`` 시 ``LongPollingViolation``. Args: seconds: 호출자가 의도한 sleep 시간 (초). Raises: LongPollingViolation: 음수, 비-int, 또는 ``MAX_SINGLE_SLEEP_SECONDS`` 초과. """ if not isinstance(seconds, int) or isinstance(seconds, bool): raise LongPollingViolation( f"sleep seconds must be plain int, got {type(seconds).__name__}" ) if seconds < 0: raise LongPollingViolation(f"negative sleep seconds forbidden: {seconds}") if seconds > MAX_SINGLE_SLEEP_SECONDS: raise LongPollingViolation( f"single sleep {seconds}s exceeds MAX_SINGLE_SLEEP_SECONDS={MAX_SINGLE_SLEEP_SECONDS}s " f"— long polling forbidden (task-2556 §9)" ) def assert_normal_wait(seconds: int) -> None: """normal wait 구간 (5~15분) 검증. 회장 § 5~15min normal wait 명시 1:1. NORMAL_WAIT_MIN_SECONDS <= seconds <= NORMAL_WAIT_MAX_SECONDS 만 허용. 범위 외 시 ``LongPollingViolation``. """ assert_sleep_allowed(seconds) if seconds < NORMAL_WAIT_MIN_SECONDS: raise LongPollingViolation( f"normal wait {seconds}s below NORMAL_WAIT_MIN_SECONDS={NORMAL_WAIT_MIN_SECONDS}s " f"— 1초 tight loop polling forbidden" ) if seconds > NORMAL_WAIT_MAX_SECONDS: raise LongPollingViolation( f"normal wait {seconds}s above NORMAL_WAIT_MAX_SECONDS={NORMAL_WAIT_MAX_SECONDS}s" ) def assert_first_timeout_not_exceeded(elapsed_seconds: int) -> None: """trigger 후 ``elapsed_seconds`` 가 ``FIRST_TIMEOUT_SECONDS`` 를 초과하면 violation. 호출자는 본 예외를 catch 후 'POSTED_BUT_NO_FRESH_EVIDENCE' 경로로 escalate 해야 한다. """ if not isinstance(elapsed_seconds, int) or isinstance(elapsed_seconds, bool): raise LongPollingViolation( f"elapsed_seconds must be plain int, got {type(elapsed_seconds).__name__}" ) if elapsed_seconds < 0: raise LongPollingViolation( f"elapsed_seconds must be non-negative, got {elapsed_seconds}" ) if elapsed_seconds > FIRST_TIMEOUT_SECONDS: raise LongPollingViolation( f"first-timeout {elapsed_seconds}s exceeds FIRST_TIMEOUT_SECONDS={FIRST_TIMEOUT_SECONDS}s " f"— scheduler must exit bot session and let scheduled/event-driven recheck resume" ) def advance_recheck(state: Any) -> PollingState: """recheck 카운터 +1. ``MAX_RECHECKS`` 초과 시 ``BotSessionExitRequired``. 호출자는 매 recheck 직후 본 함수를 호출. 본 함수는 다음 state 를 반환한다. 1 회 recheck 가 끝나면 다음 호출에서 ``BotSessionExitRequired`` 가 raise 되어 봇은 즉시 종료해야 한다. 추가 recheck 는 scheduled (cron) 또는 event-driven (webhook) 경로로만 가능. Args: state: 현재 polling state. Returns: 다음 polling state (``rechecks_done += 1``). Raises: BotSessionExitRequired: ``rechecks_done`` 이 이미 ``MAX_RECHECKS`` 에 도달. """ if not isinstance(state, PollingState): raise LongPollingViolation( f"advance_recheck expects PollingState, got {type(state).__name__}" ) if state.rechecks_done >= MAX_RECHECKS: raise BotSessionExitRequired( f"rechecks_done={state.rechecks_done} reached MAX_RECHECKS={MAX_RECHECKS} " f"— bot session must exit, next recheck via scheduler" ) return replace(state, rechecks_done=state.rechecks_done + 1) def must_exit_now(state: Any) -> bool: """봇이 즉시 exit 해야 하는 상태인지 판단. 조건 (둘 중 하나): 1. ``rechecks_done >= MAX_RECHECKS`` — 추가 recheck 권한 없음. 2. ``elapsed_seconds > FIRST_TIMEOUT_SECONDS`` — first timeout 초과. Returns: True 면 봇은 즉시 process exit. False 면 scheduled wait + 1 recheck 가능. """ if not isinstance(state, PollingState): return True if state.rechecks_done >= MAX_RECHECKS: return True if state.elapsed_seconds > FIRST_TIMEOUT_SECONDS: return True return False @dataclass(frozen=True) class SchedulerCycleBudget: """단일 scheduler cycle 의 누적 sleep 예산. 같은 process 안에서 여러 PR 를 scan 하면서 각각 normal_wait 를 호출하면 누적 sleep 이 long polling 으로 변질될 수 있다. 본 dataclass 는 single cycle 안에서 누적 sleep 상한을 ``MAX_SINGLE_SLEEP_SECONDS`` 로 강제한다. Attributes: cumulative_sleep_seconds: 본 cycle 누적 sleep. """ cumulative_sleep_seconds: int = 0 def consume_sleep_budget( budget: SchedulerCycleBudget, requested_seconds: int ) -> SchedulerCycleBudget: """sleep 예산 소비. 누적 합이 ``MAX_SINGLE_SLEEP_SECONDS`` 를 넘으면 violation. Args: budget: 현재 cycle 예산. requested_seconds: 요청 sleep 시간. Returns: 다음 budget (누적 합 += requested_seconds). Raises: LongPollingViolation: 누적 합이 상한 초과. """ assert_sleep_allowed(requested_seconds) next_total = budget.cumulative_sleep_seconds + requested_seconds if next_total > MAX_SINGLE_SLEEP_SECONDS: raise LongPollingViolation( f"cycle cumulative sleep {next_total}s would exceed " f"MAX_SINGLE_SLEEP_SECONDS={MAX_SINGLE_SLEEP_SECONDS}s — " f"split into multiple scheduled cycles" ) return replace(budget, cumulative_sleep_seconds=next_total) __all__ = [ "MAX_SINGLE_SLEEP_SECONDS", "FIRST_TIMEOUT_SECONDS", "MAX_RECHECKS", "NORMAL_WAIT_MIN_SECONDS", "NORMAL_WAIT_MAX_SECONDS", "LongPollingViolation", "BotSessionExitRequired", "PollingState", "SchedulerCycleBudget", "assert_sleep_allowed", "assert_normal_wait", "assert_first_timeout_not_exceeded", "advance_recheck", "must_exit_now", "consume_sleep_budget", ]