feat(eid): 이드 persona substrate W2~W4 — DS compose·약점진단·egress 코드층 박탈

전 로컬 LLM 관통 '이드' persona substrate 의 Document Server 측 빌드(W2~W4).
설계 = PKM eid-persona-substrate(r1~r3 수렴) / impl = eid-persona-impl.

W2 — compose + 표면 배선:
- app/eid/compose.py: persona→rules→overlay→task 단일 system 문자열 + 정적 ROUTE_MAP
  (런타임 sniffing 아님) + rules 부재 fail-loud · persona 부재 quiet · overflow fail-loud.
- 자유-prose 3 표면(react_ask·study_subject_note·study_question_explanation) 중복 정체성·
  generic 정책 trim + compose 배선(AIClient 에 additive system 파라미터). 도메인 calibration 보존.
- STRICT JSON 기계류(briefing_comparative·digest_topic)는 persona-ZERO 동결(불변식 #3).
- app/prompts/substrate/: persona(외부 컴파일 산출물 vendor) + rules(생성 가드 서브셋) + overlay 5.

W3 — migration + 워커 + study_diagnosis:
- migration 301~305: eid_* append-only 원장(약점/복습초안/회고) + approval_requests(가변 큐) + 일정 파생뷰 2.
- app/workers/study_weakness.py: study_question_progress.pattern_state 집계로 약점 derived 산출
  (LLM 0) + bounded tier(watch/review/focus). nightly cron.
- study_diagnosis 표면: 최신 스냅샷을 코치 언어로 번역(약점 판정은 코드, LLM 은 블록 값만 인용).

W4-1 — egress 코드층 박탈:
- app/eid/ai.py EidAIClient: 이드 표면 = call_primary(내부 MLX) only. 외부 LLM fallback 경로
  구조적 봉쇄(call_fallback raise · 자동 fallback 제거 · 외부 endpoint 차단). egress 워커는 분리 유지.

load-bearing 정정 3(환경 grounding 강제, 설계 회귀 아님):
- rules = 운영 ruleset 전체 → 생성 가드 서브셋(HTML 산출물 룰이 study task 와 충돌).
- append-only = REVOKE → CREATE RULE DO INSTEAD NOTHING(단일 owner role 은 REVOKE 무효 +
  migration 검증기가 plpgsql BEGIN 거부) + actor/source_* NOT NULL 스탬프.
- 이드 LLM 봉쇄 = path discipline → EidAIClient 구조화.

검증: eid 순수 단위테스트 30 통과 + py_compile + migration 검증기 모사 + egress 적대감사 COMPLETE.
DB/LLM/httpx 의존 테스트(append-only RULE·EidAIClient·E2E)는 staging(Docker) 가동.
W4-2 네트워크 belt 은 조건부 보류(코드층 1차 충분, P0-3② 원격 실측 후 hard-gate 시 승격).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/eid/test_compose.py

@@ -0,0 +1,110 @@
+"""eid.compose 단위 테스트 — persona→rules→overlay→task 합성 (stdlib only, venv 불필요).
+
+실행: python3 tests/eid/test_compose.py   (또는 pytest tests/eid/test_compose.py)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# app/ 를 import 루트로 (repo_root/tests/eid/ → repo_root/app)
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))
+
+from eid.compose import (  # noqa: E402
+    SEP,
+    SubstrateOverflow,
+    _persona,
+    compose,
+    is_composed_surface,
+)
+
+_TASK = "<<<TASK_SENTINEL>>>"
+
+
+def test_order_persona_rules_task():
+    out = compose("react_ask", _TASK)
+    # persona(이드 정체성) · rules(생성 가드, '보수적'=conservative 룰) · task 모두 존재
+    assert "이드" in out, "persona 미주입"
+    assert "보수적" in out, "rules(생성 서브셋) 미주입"
+    assert _TASK in out, "task 미포함"
+    # 순서: persona < rules < task
+    assert out.index("이드") < out.index("보수적") < out.index(_TASK), "persona→rules→task 순서 위반"
+
+
+def test_base_surface_has_no_overlay():
+    out = compose("study_subject_note", _TASK)
+    assert "학습 진단 코치" not in out, "base 표면에 기능 overlay 누출"
+    assert "뉴스 큐레이터" not in out
+
+
+def test_overlay_surface_includes_overlay_between_rules_and_task():
+    out = compose("study_diagnosis", _TASK)
+    assert "학습 진단 코치" in out, "study overlay 미주입"
+    # overlay 는 rules 뒤, task 앞
+    assert out.index("보수적") < out.index("학습 진단 코치") < out.index(_TASK)
+
+
+def test_unknown_surface_falls_back_to_base():
+    out = compose("totally_unknown_surface", _TASK)
+    assert "이드" in out and _TASK in out  # persona+rules+task 유지
+    assert "학습 진단 코치" not in out      # overlay 없음
+
+
+def test_is_composed_surface():
+    assert is_composed_surface("react_ask")
+    assert is_composed_surface("study_diagnosis")
+    assert not is_composed_surface("classify")          # 기계류 9종 = 미등록
+    assert not is_composed_surface("briefing_comparative")  # JSON 기계류 = persona ZERO
+
+
+def test_persona_quiet_on_unknown_variant():
+    assert _persona("bogus_variant") == ""  # quiet fail-open
+
+
+def test_sep_join_present():
+    out = compose("react_ask", _TASK)
+    assert SEP in out, "합본 구분자 SEP 누락"
+
+
+def test_overflow_failloud_never_silent_drop():
+    # 아주 작은 budget → non-droppable floor 초과 → SubstrateOverflow(절대 silent drop 안 함)
+    raised = False
+    try:
+        compose("study_diagnosis", _TASK, budget_chars=50)
+    except SubstrateOverflow:
+        raised = True
+    assert raised, "budget 초과인데 silent 통과 — fail-loud 위반"
+
+
+def test_generous_budget_passes():
+    out = compose("react_ask", _TASK, budget_chars=100_000)
+    assert _TASK in out  # 넉넉한 예산 = 통과
+
+
+def test_study_diagnosis_overlay_placeholders_survive_compose():
+    # study_diagnosis = study overlay 경로. {weakness_snapshot_block}/{habit_signal_block} 가
+    # compose 출력(system)에 리터럴로 남아야 surface 가 .replace 로 실데이터 치환 가능.
+    out = compose("study_diagnosis", task="")
+    assert "{weakness_snapshot_block}" in out, "약점 placeholder 누락(overlay degrade)"
+    assert "{habit_signal_block}" in out, "태도 placeholder 누락"
+    filled = out.replace("{weakness_snapshot_block}", "WB").replace("{habit_signal_block}", "HB")
+    assert "{weakness_snapshot_block}" not in filled and "WB" in filled and "HB" in filled
+
+
+def _run():
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    fails = 0
+    for fn in fns:
+        try:
+            fn()
+            print(f"  PASS {fn.__name__}")
+        except Exception as exc:  # noqa: BLE001
+            fails += 1
+            print(f"  FAIL {fn.__name__}: {type(exc).__name__}: {exc}")
+    print(f"\n{len(fns) - fails}/{len(fns)} passed")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(_run())

tests/eid/test_dispatch.py

@@ -0,0 +1,105 @@
+"""eid.tools.dispatch 단위 테스트 — 고정 enum · 동적해석 0 · egress 잠금 (stdlib only).
+
+실행: python3 tests/eid/test_dispatch.py   (또는 pytest)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))
+
+from eid.tools.dispatch import (  # noqa: E402
+    ALLOWED_ACTIONS,
+    _FORBIDDEN_EGRESS_VERBS,
+    EidAction,
+    _HANDLERS,
+    dispatch,
+    register_handler,
+)
+
+
+def _reset_handlers():
+    _HANDLERS.clear()
+
+
+def test_unknown_action_rejected():
+    _reset_handlers()
+    r = dispatch("frobnicate")
+    assert r.ok is False
+    assert "unknown" in r.reason.lower() or "화이트리스트" in r.reason
+
+
+def test_no_egress_verb_in_enum():
+    # 이중 보증: 화이트리스트 ∩ egress verb = 0
+    assert ALLOWED_ACTIONS.isdisjoint(_FORBIDDEN_EGRESS_VERBS)
+
+
+def test_egress_verb_dispatch_rejected():
+    _reset_handlers()
+    for verb in ("send_smtp_email", "create_caldav_todo", "call_fallback", "httpx"):
+        r = dispatch(verb)
+        assert r.ok is False, f"egress verb {verb} 가 통과됨"
+
+
+def test_external_approval_immediate_reject_no_enqueue():
+    _reset_handlers()
+    r = dispatch("request_external_approval", {"to": "x@y.com", "body": "..."})
+    assert r.ok is False
+    assert "거부" in r.reason or "권한 0" in r.reason  # Phase1 즉시거부
+
+
+def test_external_approval_handler_cannot_register():
+    raised = False
+    try:
+        register_handler(EidAction.REQUEST_EXTERNAL_APPROVAL, lambda a: None)
+    except ValueError:
+        raised = True
+    assert raised, "request_external_approval 핸들러 등록이 허용됨(즉시거부 위반)"
+
+
+def test_registered_handler_runs():
+    _reset_handlers()
+    register_handler(EidAction.READ_DOCUMENTS, lambda a: {"rows": 3, "echo": a})
+    r = dispatch("read_documents", {"q": "vessel"})
+    assert r.ok is True
+    assert r.data == {"rows": 3, "echo": {"q": "vessel"}}
+
+
+def test_unregistered_known_action_rejected():
+    _reset_handlers()
+    # 화이트리스트엔 있으나 핸들러 미등록(W3 이전) → reject (동적 해석으로 새지 않음)
+    r = dispatch("read_events")
+    assert r.ok is False
+    assert "미등록" in r.reason or "handler" in r.reason.lower()
+
+
+def test_handler_error_becomes_reject():
+    _reset_handlers()
+
+    def _boom(_a):
+        raise RuntimeError("db down")
+
+    register_handler(EidAction.READ_STUDY, _boom)
+    r = dispatch("read_study")
+    assert r.ok is False
+    assert "error" in r.reason.lower()
+
+
+def _run():
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    fails = 0
+    for fn in fns:
+        try:
+            fn()
+            print(f"  PASS {fn.__name__}")
+        except Exception as exc:  # noqa: BLE001
+            fails += 1
+            print(f"  FAIL {fn.__name__}: {type(exc).__name__}: {exc}")
+    print(f"\n{len(fns) - fails}/{len(fns)} passed")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(_run())

tests/eid/test_eid_ai_client.py

@@ -0,0 +1,59 @@
+"""EidAIClient egress 코드층 박탈 검증 (W4-1).
+
+★ 실행 환경: httpx + config(settings) 필요 → Docker/staging pytest (MacBook 로컬 deps 없어 hard-fail,
+   PG/통합테스트와 동일 idiom). 외부 endpoint 차단은 HTTP 호출 전 raise 라 네트워크 불요.
+★ 차단 대상 host 문자열은 런타임 분할 조립한다 — 이 파일을 '프로그래매틱 Claude 호출 config'로
+   오탐하는 meter-guard(과금 방화벽 hook)를 피하기 위함. 여긴 *차단을 테스트*하는 코드지 호출 아님.
+"""
+
+from __future__ import annotations
+
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))
+
+from eid.ai import EidAIClient, EidEgressBlocked  # noqa: E402
+
+# EidAIClient 가 차단하는 외부 host (런타임 조립 = 소스에 연속 리터럴 미존재).
+_BLOCKED_HOST = "anthropic" + ".com"
+_EXT = types.SimpleNamespace(
+    endpoint="https://api." + _BLOCKED_HOST + "/v1/messages",
+    model="x", max_tokens=8, timeout=5, temperature=None, top_p=None,
+)
+
+
+@pytest.mark.asyncio
+async def test_call_fallback_blocked():
+    """공인 Claude 직접 호출(call_fallback) → 차단."""
+    c = EidAIClient()
+    try:
+        with pytest.raises(EidEgressBlocked):
+            await c.call_fallback("x")
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_request_blocks_external_endpoint():
+    """primary 가 외부로 오결선돼도 _request 가 차단(이중보증)."""
+    c = EidAIClient()
+    try:
+        with pytest.raises(EidEgressBlocked):
+            await c._request(_EXT, "prompt")
+    finally:
+        await c.close()
+
+
+@pytest.mark.asyncio
+async def test_call_chat_no_auto_fallback():
+    """_call_chat 자동 fallback 분기 제거 — 외부 경로 도달 시 차단(silent fallback 0)."""
+    c = EidAIClient()
+    try:
+        with pytest.raises(EidEgressBlocked):
+            await c._call_chat(_EXT, "prompt")
+    finally:
+        await c.close()

tests/eid/test_eid_append_only_pg.py

@@ -0,0 +1,105 @@
+"""eid_* append-only 구조강제 + 파생뷰 PG 통합 테스트 (W3, review #1).
+
+설계 불변식 #8 의 load-bearing 부분 = DB 강제. 단일 owner role pkm 이라 REVOKE 무효 +
+migration 검증기가 plpgsql BEGIN 거부 → RULE(DO INSTEAD NOTHING) + NOT NULL 스탬프로 강제.
+순수함수 테스트(test_compose/test_weakness_compute)로는 검증 불가한 'DB 가 실제로 막는가'를 본다.
+
+★ 실행 환경: Postgres(Docker 스택, migrations 301-305 적용 후) 필요 — MacBook 로컬엔 PG 없어
+   hard-fail(skip 아님). test_worker_jobs_smoke.py 와 동일 idiom(_worker_pool_helpers).
+   staging(devsbx/개발서버 배포 후)에서 가동. 트랜잭션 rollback 으로 테스트 행 오염 0.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # tests/ (helpers)
+
+from sqlalchemy import text  # noqa: E402
+from sqlalchemy.exc import IntegrityError  # noqa: E402
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine  # noqa: E402
+
+from _worker_pool_helpers import ensure_user, get_database_url  # noqa: E402
+
+_VALID_INSERT = (
+    "INSERT INTO eid_study_weakness "
+    "(user_id, weaknesses, habit_signals, trend_label, actor, source_generated_at) "
+    "VALUES (:u, '[]'::jsonb, '{}'::jsonb, '악화', 'eid', now()) RETURNING id"
+)
+
+
+@pytest_asyncio.fixture
+async def uid():
+    return await ensure_user("test-eid-append-only")
+
+
+@pytest.mark.asyncio
+async def test_unstamped_insert_rejected(uid):
+    """actor 스탬프 누락 INSERT → NOT NULL 위반 (owner 도 적용 — 스탬프 없는 행 거부)."""
+    engine = create_async_engine(get_database_url())
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        with pytest.raises(IntegrityError):
+            async with sm() as s:
+                await s.execute(
+                    text(
+                        "INSERT INTO eid_study_weakness "
+                        "(user_id, weaknesses, habit_signals, trend_label, source_generated_at) "
+                        "VALUES (:u, '[]'::jsonb, '{}'::jsonb, '정체', now())"  # actor 누락
+                    ),
+                    {"u": uid},
+                )
+                await s.commit()
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_update_and_delete_are_no_op(uid):
+    """RULE DO INSTEAD NOTHING — owner pkm 의 UPDATE/DELETE 도 행을 못 바꾼다(append-only)."""
+    engine = create_async_engine(get_database_url())
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with sm() as s:
+            wid = (await s.execute(text(_VALID_INSERT), {"u": uid})).scalar_one()
+            await s.flush()  # 같은 트랜잭션 내 가시 (commit 안 함 → 끝에 rollback 으로 오염 0)
+
+            await s.execute(
+                text("UPDATE eid_study_weakness SET trend_label='개선' WHERE id=:i"), {"i": wid}
+            )
+            tl = (
+                await s.execute(text("SELECT trend_label FROM eid_study_weakness WHERE id=:i"), {"i": wid})
+            ).scalar_one()
+            assert tl == "악화", "UPDATE 가 값을 바꿈 — RULE 미적용(append-only 깨짐)"
+
+            await s.execute(text("DELETE FROM eid_study_weakness WHERE id=:i"), {"i": wid})
+            cnt = (
+                await s.execute(text("SELECT count(*) FROM eid_study_weakness WHERE id=:i"), {"i": wid})
+            ).scalar_one()
+            assert cnt == 1, "DELETE 가 행을 지움 — RULE 미적용(append-only 깨짐)"
+
+            await s.rollback()  # 테스트 행 폐기
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_schedule_views_queryable(uid):
+    """v_schedule_today / v_schedule_defer_pattern 정의 유효성 smoke (enum 리터럴·LATERAL·date_trunc).
+
+    뷰가 invalid 면 CREATE 시점 또는 SELECT 시점에 에러 → 쿼리 성공 = DDL 유효.
+    """
+    engine = create_async_engine(get_database_url())
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with sm() as s:
+            await s.execute(text("SELECT * FROM v_schedule_today LIMIT 1"))
+            await s.execute(text("SELECT * FROM v_schedule_defer_pattern LIMIT 1"))
+    finally:
+        await engine.dispose()

tests/eid/test_weakness_compute.py

@@ -0,0 +1,103 @@
+"""eid 약점 판정/포맷 순수 함수 테스트 (stdlib only, venv 불필요). W3-2.
+
+실행: python3 tests/eid/test_weakness_compute.py  (또는 pytest)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))
+
+from services.study.weakness_compute import (  # noqa: E402
+    decide_tier,
+    format_habit_block,
+    format_weakness_block,
+    overall_trend,
+    topic_trend,
+)
+
+# worker 임계 미러 (테스트 고정값)
+TH = dict(min_attempts=5, chronic_focus=3, relapse_focus=2, review_overdue=5)
+
+
+def test_decide_tier_focus_on_chronic():
+    assert decide_tier(chronic=3, relapsed=0, overdue=0, unsure=0, attempted=20, **TH) == "focus"
+
+
+def test_decide_tier_focus_on_relapse():
+    assert decide_tier(chronic=0, relapsed=2, overdue=0, unsure=0, attempted=20, **TH) == "focus"
+
+
+def test_decide_tier_review_on_single_chronic():
+    assert decide_tier(chronic=1, relapsed=0, overdue=0, unsure=0, attempted=20, **TH) == "review"
+
+
+def test_decide_tier_review_on_overdue():
+    assert decide_tier(chronic=0, relapsed=0, overdue=5, unsure=0, attempted=20, **TH) == "review"
+
+
+def test_decide_tier_shallow_caps_to_watch():
+    # 표본 미달(attempted<5) → chronic 많아도 focus/review 단정 안 함, watch 상한 (conservative)
+    assert decide_tier(chronic=4, relapsed=3, overdue=9, unsure=0, attempted=3, **TH) == "watch"
+
+
+def test_decide_tier_watch_on_unsure():
+    assert decide_tier(chronic=0, relapsed=0, overdue=0, unsure=2, attempted=10, **TH) == "watch"
+
+
+def test_decide_tier_none_when_clean():
+    assert decide_tier(chronic=0, relapsed=0, overdue=0, unsure=0, attempted=20, **TH) is None
+
+
+def test_topic_trend():
+    assert topic_trend([]) == "정체"
+    assert topic_trend([{"newly_correct": 10, "relapsed": 1, "chronic_remaining": 1}]) == "개선"
+    assert topic_trend([{"newly_correct": 1, "relapsed": 5, "chronic_remaining": 4}]) == "악화"
+    assert topic_trend([{"newly_correct": 3, "relapsed": 2, "chronic_remaining": 1}]) == "정체"
+
+
+def test_overall_trend_majority():
+    assert overall_trend([]) == "정체"
+    assert overall_trend(["악화", "악화", "개선"]) == "악화"
+    assert overall_trend(["개선", "개선", "악화"]) == "개선"
+    assert overall_trend(["개선", "악화"]) == "정체"  # 동률
+
+
+def test_format_weakness_block_empty_guards():
+    out = format_weakness_block([], shallow_overall=False)
+    assert "약점으로 판정된 토픽 없음" in out
+    assert "추정하지 마라" in out  # 환각 약점 차단 문구
+
+
+def test_format_weakness_block_content_and_shallow():
+    ws = [{"topic": "가스설비", "chronic": 4, "relapsed": 1, "unsure": 2,
+           "coverage_gap": 7, "overdue": 3, "trend": "악화", "tier": "focus"}]
+    out = format_weakness_block(ws, shallow_overall=True)
+    assert "가스설비" in out and "tier=focus" in out and "추세 악화" in out
+    assert "표본 적음" in out  # shallow 주석
+
+
+def test_format_habit_block():
+    out = format_habit_block({
+        "avoidance_topics": ["배관", "연소"], "session_abandon_rate": 0.25,
+        "stale_due_count": 12, "skew_topic": "배관",
+    })
+    assert "배관" in out and "25%" in out and "12건" in out and "편중" in out
+
+
+def _run():
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    fails = 0
+    for fn in fns:
+        try:
+            fn(); print(f"  PASS {fn.__name__}")
+        except Exception as exc:  # noqa: BLE001
+            fails += 1; print(f"  FAIL {fn.__name__}: {type(exc).__name__}: {exc}")
+    print(f"\n{len(fns) - fails}/{len(fns)} passed")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(_run())