hyungi_document_server/app/eid/compose.py

"""이드 substrate compose — persona → rules → overlay → task 단일 system 문자열.

설계 정본 : PKM plans/2026-06-05-eid-persona-substrate-plan.html (eid-persona-substrate, r1~r3 수렴)
구현 plan : plans/2026-06-07-eid-persona-impl-plan.html (W2-1)
불변식    : memory project_eid_persona_substrate (load-bearing 9건)

핵심 불변식 (바꾸지 말 것 — 위반 = 설계 회귀):
  #3 "강력하게" = 출력계약 경계(균질주입 아님). 자유-prose 표면 = persona ON,
     STRICT JSON 기계류 = persona ZERO. 판정 = 정적 ROUTE_MAP(런타임 sniffing 아님).
  #4 합본 = persona → rules → overlay → task. rules 는 합본의 *명시 항*(compose 가 반드시 끼움)
     → 'rules 부재 = fail-loud' 성립. 충돌 시 rules > persona, overlay ≤ rules.
     persona 부재 = quiet fail-open / rules 부재 = fail-loud(degraded 배너 + 로그).
  #2 overlay 는 delta-only. injection 방어는 공통 rules(rules.md)에 있음(overlay 아님, never-dropped).

스코프: 사용자대면 자유-prose 표면만. STRICT JSON 기계류 9종은 ROUTE_MAP 부재 → compose 우회(task-only).

의존성: stdlib only (DB·yaml·LLM 불필요). 입력 = app/prompts/substrate/ 의 vendored 아티팩트.
"""

from __future__ import annotations

import logging
from functools import lru_cache
from pathlib import Path

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

# vendored 아티팩트 (sync = app/prompts/substrate/README.md)
_SUBSTRATE_DIR = Path(__file__).resolve().parent.parent / "prompts" / "substrate"
_OVERLAY_DIR = _SUBSTRATE_DIR / "overlays"

# 합본 구분자 — MLX 다중 system role 위험 회피용 단일 문자열 join (설계 0-3)
SEP = "\n\n---\n\n"

# variant → persona 아티팩트 파일명. 26B/27B = full, 4B = compact.
_PERSONA_FILES = {"full": "persona.full.md", "compact": "persona.compact.md"}

# rules 미주입 시 degraded 배너 (fail-loud — silent 빈문자열 금지, 불변식 #4)
_RULES_DEGRADED = (
    "[substrate-degraded: 운영 규칙(rules) 미주입 — 안전·정책 가드 없이 동작 중. "
    "app/prompts/substrate/rules.md 부재. 관리자 확인 필요.]"
)

# ── 정적 ROUTE_MAP (surface → overlay + variant). 런타임 출력 sniffing 아님(불변식 #3). ──
# overlay=None → 자유-prose 표면(persona + rules + task, 기능 overlay 없음).
# overlay name → 미래 active eid 표면(W3+ 배선). variant = persona 변형(현재 전부 26B/27B = full).
# 미등록 surface(.get None) → base(persona + rules + task) + 가시 로그.
_ROUTE: dict[str, dict] = {
    # W2-2 wire 대상 — 자유-prose, 기능 overlay 없음(base)
    "react_ask":                  {"overlay": None, "variant": "full"},
    "study_subject_note":         {"overlay": None, "variant": "full"},
    "study_question_explanation": {"overlay": None, "variant": "full"},
    # 미래 active eid 표면 — 기능 overlay (W3+ 에서 호출 배선)
    "study_diagnosis":            {"overlay": "study",    "variant": "full"},
    "document_brief":             {"overlay": "document", "variant": "full"},
    "news_brief":                 {"overlay": "news",     "variant": "full"},
    "recap_brief":                {"overlay": "recap",    "variant": "full"},
    "schedule_brief":             {"overlay": "schedule", "variant": "full"},
}


class SubstrateOverflow(RuntimeError):
    """non-droppable floor 가 모델 budget 초과 — fail-loud(26B 에스컬레이트), 절대 silent drop 안 함."""


@lru_cache(maxsize=8)
def _read(path_str: str) -> str | None:
    """파일 읽기(캐시). 부재 = None (호출부가 quiet/loud 결정)."""
    p = Path(path_str)
    if not p.is_file():
        return None
    return p.read_text(encoding="utf-8").strip()


def _persona(variant: str) -> str:
    """persona 변형 로드. 부재 = quiet fail-open(빈 문자열) — voice 는 cosmetic(불변식 #4)."""
    fname = _PERSONA_FILES.get(variant)
    if fname is None:
        logger.debug("eid.compose: unknown persona variant %r → quiet skip", variant)
        return ""
    text = _read(str(_SUBSTRATE_DIR / fname))
    if text is None:
        logger.debug("eid.compose: persona %r absent → quiet fail-open", fname)
        return ""
    return text


def _rules() -> str:
    """rules 로드. 부재 = fail-loud(degraded 배너 + error 로그) — 정책은 silent 누락 금지(불변식 #4)."""
    text = _read(str(_SUBSTRATE_DIR / "rules.md"))
    if text is None:
        logger.error(
            "eid.compose: rules.md ABSENT — substrate degraded (안전·정책 가드 없이 동작). "
            "app/prompts/substrate/rules.md 확인 필요."
        )
        return _RULES_DEGRADED
    return text


def _overlay(name: str | None) -> str:
    """기능 overlay 로드. name=None → 빈 문자열(base). 미존재 파일 = fail-loud(error 로그 + 빈)."""
    if name is None:
        return ""
    text = _read(str(_OVERLAY_DIR / f"{name}.txt"))
    if text is None:
        logger.error("eid.compose: overlay %r 파일 부재 → base 로 degrade", name)
        return ""
    return text


def is_composed_surface(surface: str) -> bool:
    """이 surface 가 ROUTE_MAP 에 등록된 compose 대상인가(= persona 주입 표면인가)."""
    return surface in _ROUTE


def compose(surface: str, task: str, *, variant: str | None = None,
            budget_chars: int | None = None) -> str:
    """persona → rules → overlay → task 단일 system 문자열 합성.

    surface     : 정적 ROUTE_MAP 키. 미등록이면 base(persona+rules+task) + 가시 로그.
    task        : 표면 고유 지시(기존 prompt txt 본문). 합본의 마지막 항.
    variant     : persona 변형 override. None = ROUTE_MAP 의 variant(기본 full).
    budget_chars: 모델 system 예산(char). None = 무제한(26B/27B 경로). 설정 시 non-droppable
                  floor(persona+rules+overlay) 초과면 SubstrateOverflow(fail-loud, 절대 silent drop X).

    반환: SEP 로 join 된 system 문자열. 빈 항(persona 부재 등)은 join 에서 제외.
    """
    route = _ROUTE.get(surface)
    if route is None:
        logger.info(
            "eid.compose: surface %r ROUTE_MAP 미등록 → base(persona+rules+task)", surface
        )
        v = variant or "full"
        overlay_name = None
    else:
        v = variant or route["variant"]
        overlay_name = route["overlay"]

    persona = _persona(v)
    rules = _rules()                 # 항상 비-빈(degraded 배너라도) → 합본의 명시 항 보장
    overlay = _overlay(overlay_name)

    # non-droppable floor = persona + rules + overlay (task 제외). budget 초과 = fail-loud.
    if budget_chars is not None:
        floor = len(SEP.join(p for p in (persona, rules, overlay) if p))
        if floor > budget_chars:
            logger.error(
                "eid.compose: non-droppable floor %d char > budget %d (surface=%r, variant=%r) "
                "→ fail-loud, 26B 에스컬레이트 필요(silent drop 안 함)",
                floor, budget_chars, surface, v,
            )
            raise SubstrateOverflow(
                f"floor {floor} > budget {budget_chars} for surface={surface!r} variant={v!r}"
            )

    parts = [persona, rules, overlay, task]
    return SEP.join(p for p in parts if p)


def clear_cache() -> None:
    """vendored 아티팩트 sync 후 재로드용(1회 캐시 불변식). 프로세스 재시작 대안."""
    _read.cache_clear()