"""이드 액션 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)