{#if revealed && current.evidence?.length && current.evidence[0].snippet}
-
+
근거
-
{current.evidence[0].snippet}
+
{@html renderMathMarkdown(current.evidence[0].snippet)}
{:else if revealed}
확정 풀이에서 추출한 카드 — 별도 인용 없음
diff --git a/migrations/301_eid_study_weakness.sql b/migrations/301_eid_study_weakness.sql
new file mode 100644
index 0000000..38c74c4
--- /dev/null
+++ b/migrations/301_eid_study_weakness.sql
@@ -0,0 +1,40 @@
+-- 301_eid_study_weakness.sql
+-- 이드 학습 약점 스냅샷 (append-only derived-fact). eid_study_weakness 워커가 study_question_progress
+-- + study_quiz_sessions 집계로 산출(LLM 0). study_diagnosis 표면이 최신 행을 읽어 코치 발화.
+--
+-- ★ append-only 구조강제 (project_eid_persona_substrate 불변식 #8) — 2중:
+-- (1) INSERT 스탬프 누락 거부: actor·source_generated_at = NOT NULL·DEFAULT 없음
+-- → 스탬프 없는 INSERT 를 DB 가 거부. NOT NULL 은 owner 포함 모든 role 에 적용(role 독립).
+-- (2) UPDATE/DELETE 차단: CREATE RULE ... DO INSTEAD NOTHING → 행 불변(owner·superuser 독립).
+--
+-- ★ 설계 원안 'REVOKE UPDATE,DELETE' 정정(load-bearing): 단일 DB role `pkm` 이 테이블 OWNER 라
+-- REVOKE 가 무효(owner 는 GRANT/REVOKE 우회). plpgsql trigger(RAISE)는 migration 검증기가
+-- 본문의 BEGIN 키워드를 거부(_validate_sql_content)해 불가. → RULE 이 owner 독립 + 검증기 통과하는
+-- 유일한 구조 enforcement(silent no-op, 행은 구조적으로 불변). 별도 read-only role 미존재.
+--
+-- ★ '현재' 스냅샷 = 최신 created_at 행(WHERE status='active'). 상태전이 UPDATE 없음(append-only).
+-- dispute = status='disputed' + supersedes_id 로 특정 스냅샷 무효화(새 INSERT). 표면이 disputed 제외.
+--
+-- runner = exec_driver_sql(simple protocol) → multi-statement 처리(001_initial_schema 선례, 18 stmt).
+-- BEGIN/COMMIT/ROLLBACK 없음(검증기 통과). CREATE RULE 은 IF NOT EXISTS 미지원 → OR REPLACE 로 idempotent.
+
+CREATE TABLE IF NOT EXISTS eid_study_weakness (
+ id BIGSERIAL PRIMARY KEY,
+ user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ weaknesses JSONB NOT NULL, -- [{topic_id,topic,chronic,relapsed,leech,coverage_gap,unsure,overdue,trend,tier}]
+ habit_signals JSONB NOT NULL, -- {avoidance_topics,session_abandon_rate,stale_due_count,skew_topics}
+ trend_label VARCHAR(20) NOT NULL, -- overall 개선|정체|악화 (코드 산출)
+ sample_attempts INTEGER NOT NULL DEFAULT 0,
+ is_shallow_sample BOOLEAN NOT NULL DEFAULT false,
+ status VARCHAR(20) NOT NULL DEFAULT 'active', -- 'active' | 'disputed'(dispute marker)
+ supersedes_id BIGINT REFERENCES eid_study_weakness(id) ON DELETE SET NULL,
+ actor VARCHAR(20) NOT NULL, -- 스탬프(no default) = 'eid'
+ source_generated_at TIMESTAMPTZ NOT NULL, -- 스탬프(no default) = 집계 시점
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE OR REPLACE RULE eid_study_weakness_no_update AS ON UPDATE TO eid_study_weakness DO INSTEAD NOTHING;
+CREATE OR REPLACE RULE eid_study_weakness_no_delete AS ON DELETE TO eid_study_weakness DO INSTEAD NOTHING;
+
+CREATE INDEX IF NOT EXISTS idx_eid_weakness_user_current
+ ON eid_study_weakness (user_id, created_at DESC) WHERE status = 'active';
diff --git a/migrations/302_eid_review_set_draft.sql b/migrations/302_eid_review_set_draft.sql
new file mode 100644
index 0000000..84cddfd
--- /dev/null
+++ b/migrations/302_eid_review_set_draft.sql
@@ -0,0 +1,26 @@
+-- 302_eid_review_set_draft.sql
+-- 이드 복습세트 초안 (append-only derived-fact). 워커가 약점 스냅샷에서 권장 복습세트를 '제안'만 한다.
+-- study overlay 항목6: "복습세트를 실제 복습 큐에 편성은 자율로 못 한다 — 초안만 제시, 사용자 1클릭".
+-- 실제 편성(study_question_progress.due_at 편집)은 별도 T2 액션 — 이 draft 는 불변 제안 기록.
+--
+-- append-only 구조강제(=301 동일): actor·source_generated_at NOT NULL no-default(스탬프) + RULE(불변).
+-- 상태전이 없음 — '현재 제안' = 최신 created_at. 새 제안은 supersedes_id 로 이전 것 가리킴(새 INSERT).
+-- question_ids = ordered list[int] snapshot (study_quiz_sessions.question_ids 패턴, junction 안 씀).
+
+CREATE TABLE IF NOT EXISTS eid_review_set_draft (
+ id BIGSERIAL PRIMARY KEY,
+ user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ study_topic_id BIGINT REFERENCES study_topics(id) ON DELETE CASCADE, -- nullable = cross-topic 세트
+ question_ids JSONB NOT NULL, -- ordered list[int]
+ reason VARCHAR(40) NOT NULL, -- chronic | relapse | coverage | overdue
+ actor VARCHAR(20) NOT NULL, -- 스탬프 = 'eid'
+ source_weakness_id BIGINT REFERENCES eid_study_weakness(id) ON DELETE SET NULL,
+ source_generated_at TIMESTAMPTZ NOT NULL, -- 스탬프
+ supersedes_id BIGINT REFERENCES eid_review_set_draft(id) ON DELETE SET NULL,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE OR REPLACE RULE eid_review_set_draft_no_update AS ON UPDATE TO eid_review_set_draft DO INSTEAD NOTHING;
+CREATE OR REPLACE RULE eid_review_set_draft_no_delete AS ON DELETE TO eid_review_set_draft DO INSTEAD NOTHING;
+
+CREATE INDEX IF NOT EXISTS idx_eid_review_draft_user ON eid_review_set_draft (user_id, created_at DESC);
diff --git a/migrations/303_eid_weekly_recap.sql b/migrations/303_eid_weekly_recap.sql
new file mode 100644
index 0000000..fab5ca5
--- /dev/null
+++ b/migrations/303_eid_weekly_recap.sql
@@ -0,0 +1,27 @@
+-- 303_eid_weekly_recap.sql
+-- 이드 주간 회고 카드 (append-only derived-fact). 회고 워커(scaffold, 미배선 — W4/Phase2)가 산출.
+-- recap overlay: 'T1 write 자율 eid_weekly_recap(append-only)'. 미결 액션아이템 open/done UPDATE 는
+-- events 측(가변)이지 이 카드가 아님 — 카드 자체는 불변 스냅샷.
+-- 현재는 통합 migration 의 scaffold 테이블(dispatch enum WRITE_WEEKLY_RECAP 의 write target 예약).
+--
+-- append-only 구조강제(=301 동일): 스탬프 NOT NULL no-default + RULE(불변). '현재' = 최신 created_at.
+
+CREATE TABLE IF NOT EXISTS eid_weekly_recap (
+ id BIGSERIAL PRIMARY KEY,
+ user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ period_start DATE NOT NULL,
+ period_end DATE NOT NULL,
+ recap JSONB NOT NULL, -- {activity_summary, open_action_items:[event_id], recurring_topics}
+ trend_label VARCHAR(20),
+ status VARCHAR(20) NOT NULL DEFAULT 'active', -- 'active' | 'disputed'
+ supersedes_id BIGINT REFERENCES eid_weekly_recap(id) ON DELETE SET NULL,
+ actor VARCHAR(20) NOT NULL, -- 스탬프 = 'eid'
+ source_generated_at TIMESTAMPTZ NOT NULL, -- 스탬프
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE OR REPLACE RULE eid_weekly_recap_no_update AS ON UPDATE TO eid_weekly_recap DO INSTEAD NOTHING;
+CREATE OR REPLACE RULE eid_weekly_recap_no_delete AS ON DELETE TO eid_weekly_recap DO INSTEAD NOTHING;
+
+CREATE INDEX IF NOT EXISTS idx_eid_recap_user_current
+ ON eid_weekly_recap (user_id, created_at DESC) WHERE status = 'active';
diff --git a/migrations/304_approval_requests.sql b/migrations/304_approval_requests.sql
new file mode 100644
index 0000000..899c7f8
--- /dev/null
+++ b/migrations/304_approval_requests.sql
@@ -0,0 +1,24 @@
+-- 304_approval_requests.sql
+-- 외부 전송 승인 큐 (★ 가변 workflow queue — append-only 아님). 설계 3-4 명시 카브아웃:
+-- "approval_requests 는 status 를 pending→approved 로 바꾸는 가변 state 라 eid_* 불변 REVOKE/RULE 대상 아님".
+-- → 여기엔 RULE(append-only) 안 건다. status 전이(UPDATE) 허용.
+--
+-- ★ Phase1 현재: app/eid/tools/dispatch.py 의 request_external_approval = 즉시 거부(INSERT 0).
+-- dispatcher 워커(유일 egress 집행)는 Phase3. 이 테이블은 그때까지 scaffold(빈 상태).
+-- ★ payload 는 고정 템플릿 슬롯만(free-form 금지) — app 층이 request_type 별 화이트리스트 검증.
+-- 승인 UI 는 전송 body 전문 diff 노출. 불변 결정 원장이 필요하면 별도 append-only approval_events(Phase3).
+
+CREATE TABLE IF NOT EXISTS approval_requests (
+ id BIGSERIAL PRIMARY KEY,
+ user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ request_type VARCHAR(40) NOT NULL, -- 고정 템플릿 슬롯 타입(app 화이트리스트)
+ payload JSONB NOT NULL, -- 고정 템플릿 슬롯만
+ status VARCHAR(20) NOT NULL DEFAULT 'pending', -- pending | approved | rejected (전이 허용)
+ requester VARCHAR(20) NOT NULL, -- 'eid'
+ decided_by VARCHAR(40),
+ decided_at TIMESTAMPTZ,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_approval_requests_status ON approval_requests (status, created_at);
diff --git a/migrations/305_eid_schedule_views.sql b/migrations/305_eid_schedule_views.sql
new file mode 100644
index 0000000..8873aae
--- /dev/null
+++ b/migrations/305_eid_schedule_views.sql
@@ -0,0 +1,33 @@
+-- 305_eid_schedule_views.sql
+-- 이드 일정(schedule_brief, 미래 surface) 파생뷰 2. 신규 schedule 테이블 0 — events/events_history 재활용.
+-- quadrant(중요×긴급)·D-N 정렬은 app 층(schedule overlay). 뷰는 raw 입력 필드 + today/defer 집계만.
+-- CREATE VIEW 선례 = 010_soft_delete / 283_corpus_chunks. BEGIN/COMMIT 없음.
+--
+-- v_schedule_today: 오늘(Asia/Seoul local day) 활성 일정. active 필터 = events.py:list_today reference.
+-- today 경계 = Seoul 자정→UTC 변환(date_trunc ... AT TIME ZONE 왕복). LATERAL 로 1회 계산.
+-- v_schedule_defer_pattern: events_history change_kind IN(defer,reschedule) 를 event_id 별 COUNT.
+-- '반복 미룸' 임계 3회+ (schedule overlay 판단근거 #5). reactivate 는 제외.
+
+CREATE OR REPLACE VIEW v_schedule_today AS
+SELECT e.id, e.user_id, e.title, e.kind, e.status, e.priority,
+ e.due_at, e.start_at, e.end_at, e.started_at, e.defer_until, e.project_tag
+FROM events e
+CROSS JOIN LATERAL (
+ SELECT (date_trunc('day', now() AT TIME ZONE 'Asia/Seoul') AT TIME ZONE 'Asia/Seoul') AS lo
+) b
+WHERE (e.status IN ('inbox','next','scheduled','in_progress')
+ OR (e.status = 'deferred' AND e.defer_until IS NOT NULL AND e.defer_until <= now()))
+ AND (
+ (e.kind = 'task' AND e.due_at >= b.lo AND e.due_at < b.lo + interval '1 day')
+ OR (e.kind = 'calendar_event' AND e.start_at >= b.lo AND e.start_at < b.lo + interval '1 day')
+ OR (e.kind = 'activity_log' AND e.started_at >= b.lo AND e.started_at < b.lo + interval '1 day')
+ );
+
+CREATE OR REPLACE VIEW v_schedule_defer_pattern AS
+SELECT eh.event_id,
+ COUNT(*)::int AS defer_reschedule_count,
+ MAX(eh.changed_at) AS last_changed_at,
+ (COUNT(*) >= 3) AS is_repeat_defer
+FROM events_history eh
+WHERE eh.change_kind IN ('defer','reschedule')
+GROUP BY eh.event_id;
diff --git a/tests/eid/__init__.py b/tests/eid/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/eid/test_compose.py b/tests/eid/test_compose.py
new file mode 100644
index 0000000..3042b06
--- /dev/null
+++ b/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 = "<<
>>"
+
+
+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())
diff --git a/tests/eid/test_dispatch.py b/tests/eid/test_dispatch.py
new file mode 100644
index 0000000..db8c1b9
--- /dev/null
+++ b/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())
diff --git a/tests/eid/test_eid_ai_client.py b/tests/eid/test_eid_ai_client.py
new file mode 100644
index 0000000..1a2674d
--- /dev/null
+++ b/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()
diff --git a/tests/eid/test_eid_append_only_pg.py b/tests/eid/test_eid_append_only_pg.py
new file mode 100644
index 0000000..bb4928a
--- /dev/null
+++ b/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()
diff --git a/tests/eid/test_weakness_compute.py b/tests/eid/test_weakness_compute.py
new file mode 100644
index 0000000..7d47967
--- /dev/null
+++ b/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())