hyungi_document_server/app/eid/tools/dispatch.py

"""이드 액션 dispatch — 고정 enum, 동적 해석 0 (egress 코드층 능력박탈 1차).

설계 정본 : PKM plans/2026-06-05-eid-persona-substrate-plan.html §3-1 (고정 dispatch 불변식)
구현 plan : plans/2026-06-07-eid-persona-impl-plan.html (W2-4)
불변식    : memory project_eid_persona_substrate #5, #8

핵심 (바꾸지 말 것 — 위반 = egress 잠금 회귀):
  - LLM 이 낸 action 명을 *닫힌 enum* 에 대조. getattr/eval/동적 import/setattr 0. 미지 = reject.
    ReAct 가 action 을 *고르는* 것 자체는 허용(루프 본질) — 막는 건 *이름의 동적 해석*.
  - enum 에 egress verb(send_smtp_email/create_caldav_todo/httpx/call_fallback) *미포함* —
    이중 보증(import-time assert 로 강제). 같은 컨테이너에 egress 함수가 import 돼 있어도
    이드는 그 이름을 dispatch 할 수 없다.
  - 핸들러 = 정적 dict 매핑(register_handler 로 명시 등록). 동적 발견 아님. 미등록 = reject.
  - T3 external = 권한 0. Phase1 request_external_approval = *즉시 거부*(INSERT 안 함).
    dispatcher 없는 상태에서 pending 무한적재 + 소비 안 되는 큐 노출 회피. pending INSERT 는
    dispatcher 있는 Phase3 부터(W2-4 'INSERT만' ↔ D-2 침묵 불일치 해소).

의존성: stdlib only. 실제 read/write 핸들러는 W3(eid_* migration) 후 register_handler 로 주입.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable

logger = logging.getLogger("eid.dispatch")


class EidAction(str, Enum):
    """이드 호출 가능 액션 화이트리스트. *내부 액션만* — egress verb 절대 미포함.

    Tier (project_eid_persona_substrate #8):
      T0 read     = 자율 / T1 write-derived = 자율(append-only) / T2 action = 조건부(1클릭)
      T3 external = 권한 0 (approval_requests 큐만, Phase1 = 즉시 거부)
    """

    # ── T0 read (자율) ──
    READ_DOCUMENTS = "read_documents"
    READ_EVENTS = "read_events"
    READ_STUDY = "read_study"
    READ_NEWS = "read_news"
    # ── T1 write-derived (append-only, 자율) — 핸들러는 W3(eid_* 테이블) 후 ──
    WRITE_STUDY_WEAKNESS = "write_study_weakness"
    WRITE_REVIEW_SET_DRAFT = "write_review_set_draft"
    WRITE_WEEKLY_RECAP = "write_weekly_recap"
    # ── T2 conditional (사용자 1클릭 승인 후) ──
    SCHEDULE_REVIEW_SET = "schedule_review_set"
    # ── T3 external = 권한 0. Phase1 = 즉시 거부(아래 dispatch 특수 분기) ──
    REQUEST_EXTERNAL_APPROVAL = "request_external_approval"


ALLOWED_ACTIONS: frozenset[str] = frozenset(a.value for a in EidAction)

# egress verb 블랙리스트 — enum 에 *절대* 없어야 함(이중 보증). 같은 프로세스에 import 된
# core/utils.send_smtp_email·create_caldav_todo / httpx / ai.client.call_fallback 등을 가리킴.
_FORBIDDEN_EGRESS_VERBS: frozenset[str] = frozenset({
    "send_smtp_email", "create_caldav_todo", "call_fallback",
    "httpx", "http_get", "http_post", "fetch_url", "fetch",
    "webhook", "push", "send_email", "upload", "post_external",
})

# import-time 단언: 화이트리스트와 egress verb 교집합 = 0 (불변식 #5 이중 보증)
assert not (ALLOWED_ACTIONS & _FORBIDDEN_EGRESS_VERBS), (
    "eid dispatch enum 에 egress verb 포함 — 불변식 #5 위반: "
    f"{sorted(ALLOWED_ACTIONS & _FORBIDDEN_EGRESS_VERBS)}"
)


@dataclass
class DispatchResult:
    ok: bool
    action: str
    reason: str = ""
    data: Any = None
    meta: dict = field(default_factory=dict)


# 정적 핸들러 매핑 — action(str) → callable(args:dict) → data. getattr/동적 X.
# 부팅 시 register_handler 로 명시 등록(W3+). 미등록 action = reject(핸들러 없음).
_HANDLERS: dict[str, Callable[[dict], Any]] = {}


def register_handler(action: EidAction, fn: Callable[[dict], Any]) -> None:
    """핸들러 정적 등록(명시). 동적 발견 아님. egress 분기는 등록 불가(아래 가드)."""
    if action.value in _FORBIDDEN_EGRESS_VERBS:  # 도달 불가(enum 가드)이나 방어적 이중확인
        raise ValueError(f"egress verb 핸들러 등록 거부: {action.value}")
    if action == EidAction.REQUEST_EXTERNAL_APPROVAL:
        raise ValueError("request_external_approval 은 Phase1 즉시거부 — 핸들러 등록 불가")
    _HANDLERS[action.value] = fn


def _reject(action: str, reason: str) -> DispatchResult:
    logger.warning("eid.dispatch REJECT action=%r reason=%s", action, reason)
    return DispatchResult(ok=False, action=action, reason=reason)


def dispatch(action: str, args: dict | None = None) -> DispatchResult:
    """이드가 고른 action 을 *고정 분기*로 실행. 동적 이름 해석 0.

    1) 닫힌 enum 화이트리스트 대조 — 미지 = reject (getattr/eval 안 함).
    2) T3 external Phase1 = 즉시 거부(INSERT 안 함).
    3) 정적 핸들러 dict lookup — 미등록 = reject (W3 이전엔 read/write 핸들러 부재).
    """
    args = args or {}

    # 1) allowlist (닫힌 enum). 동적 해석 없이 멤버십만 본다.
    if action not in ALLOWED_ACTIONS:
        return _reject(action, "unknown action — eid enum 화이트리스트 외 (동적 해석 거부)")

    # 2) T3 external = 권한 0. Phase1 즉시 거부(적재 안 함).
    if action == EidAction.REQUEST_EXTERNAL_APPROVAL.value:
        return _reject(
            action,
            "external egress = 권한 0. Phase1: 승인큐 비활성 → 거부(pending 적재 안 함). "
            "외부 전송은 사용자(요청자≠집행자) 경유.",
        )

    # 3) 정적 핸들러 lookup (dict — getattr 아님). 미등록 = reject.
    fn = _HANDLERS.get(action)
    if fn is None:
        return _reject(action, "handler 미등록 (W3 eid_* 핸들러 주입 이전)")

    try:
        data = fn(args)
    except Exception as exc:  # 핸들러 오류 = reject(loud), 다른 분기로 새지 않음
        logger.exception("eid.dispatch handler error action=%r", action)
        return _reject(action, f"handler error: {type(exc).__name__}")

    return DispatchResult(ok=True, action=action, data=data)