2.5 QA: ai-service pytest 스위트 승격 + 자율성 플로우(L0~L2) 통합 테스트

ai-service:
- escalation_filter/retrieve_style/generation/main(FastAPI)의 ad-hoc
  TestClient 검증을 ai-service/tests/ 정식 pytest 스위트로 승격(34개).
  Gemini 호출은 mock, retrieve_style은 overlap이 recency를 항상
  이긴다는 것(과거 버그 재발 방지)까지 포함
- on_event(deprecated) -> lifespan 컨텍스트 매니저로 교체

core-backend: 자율성 플로우(L0->L1->L2) 통합 테스트를 쓰려면 실제 분기
로직이 있어야 해서, roadmap.md §2.2에서 미결로 남아있던 자율성 엔진
오케스트레이션 최소 버전을 이번에 구현:
- PATCH /users/:id/twin-settings -- 자율성 레벨 변경 (기본값 L0)
- 트윈 발송 시 에스컬레이션 통과 후 레벨 확인: L0는 항상 차단, L1은
  approved:true 필요, L2는 화이트리스트 매칭 시 즉시 자동발송·매칭
  없으면 L1과 동일하게 승인 필요. 에스컬레이션은 레벨/화이트리스트/
  승인 여부와 무관하게 항상 우선(테스트로 확인)
- 화이트리스트 매칭은 v1 최소 구현(전역 키워드 매칭, 상대별 예외는
  아직) -- 대화방↔연락처 연결이 모델링되지 않아 보류, README에 명시

온보딩·채팅·설정 수동 QA는 Flutter 클라이언트가 없어 이 환경에서는
보류, roadmap.md에 근거 남김

ai-service/README.md

@@ -16,6 +16,19 @@ export GEMINI_API_KEY=...   # 또는 이 디렉토리에 .env 파일
 uvicorn app.main:app --reload --port 8001
 ```
 
+## 테스트
+
+```bash
+pip install -r requirements-dev.txt
+pytest tests/ -v
+```
+
+`escalation_filter.py`(SELFTEST_CASES 승격 + 추가 케이스)·`retrieve_style.py`(overlap이 recency를
+항상 이긴다는 것, 동점일 때만 recency가 tie-break한다는 것)·`generation.py`(escalate/no_key/ok 세
+경로, Gemini 호출은 mock)·`main.py`(`/health`·`/escalate/check`·`/draft` 전부)를 `tests/`의 정식
+pytest 스위트로 승격함 (`docs/roadmap.md` Phase 1 §2.5). 이전엔 ad-hoc TestClient 스크립트로만
+확인했던 것들.
+
 ## API
 
 ### `POST /escalate/check`

ai-service/app/main.py

@@ -1,3 +1,4 @@
+from contextlib import asynccontextmanager
 from typing import List, Optional
 
 from fastapi import FastAPI
@@ -7,12 +8,14 @@ from .escalation_filter import check as check_escalation
 from .generation import draft_reply, last_incoming_text, load_dotenv_if_present
 from .retrieve_style import retrieve as retrieve_style_examples
 
-app = FastAPI(title="분신 AI service")
 
-
+@asynccontextmanager
-@app.on_event("startup")
+async def lifespan(app: FastAPI):
-def startup():
     load_dotenv_if_present()
+    yield
+
+
+app = FastAPI(title="분신 AI service", lifespan=lifespan)
 
 
 @app.get("/health")

ai-service/pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+pythonpath = .

ai-service/requirements-dev.txt

@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest>=8.0.0

ai-service/tests/test_escalation_filter.py

@@ -0,0 +1,48 @@
+"""Promotes escalation_filter.py's SELFTEST_CASES (poc/tone-corpus/escalation_filter.py,
+identical logic here) into a real pytest suite -- roadmap.md Phase 1 §2.5."""
+import pytest
+
+from app.escalation_filter import check
+
+SELFTEST_CASES = [
+    ("계좌로 3만원만 보내줘", True, "금전"),
+    ("그 카페 계좌번호 좀 알려줄래", True, "금전"),
+    ("그럼 내일 3시 맞지?", True, "약속 확정"),
+    ("약속 시간 확정하자, 언제가 좋아", True, "약속 확정"),
+    ("나 요즘 너무 힘들어서 죽고싶다는 생각이 들어", True, "감정적으로 무거운 주제"),
+    ("우리 어제 왜 그렇게 싸웠어", True, "감정적으로 무거운 주제"),
+    ("오늘 저녁에 뭐 먹을래?", False, ""),
+    ("크라비 여행 가보고 싶어", False, ""),
+    ("고등학교 학점제가 뭔지 설명해줄 수 있어?", False, ""),
+    ("이 영화 재밌었어? 나도 보고싶다", False, ""),
+]
+
+
+@pytest.mark.parametrize("text,expected_escalate,expected_reason", SELFTEST_CASES)
+def test_selftest_cases(text, expected_escalate, expected_reason):
+    result = check(text)
+    assert result.escalate == expected_escalate
+    if expected_escalate:
+        assert result.reason == expected_reason
+    else:
+        assert result.reason == ""
+
+
+def test_money_wins_over_no_match_when_both_patterns_could_apply():
+    # 금전 패턴이 먼저 검사되므로 금전+약속이 섞여도 이유는 금전으로 고정된다.
+    result = check("계좌번호 알려주면 내일 3시 맞지?")
+    assert result.escalate is True
+    assert result.reason == "금전"
+
+
+def test_empty_text_never_escalates():
+    result = check("")
+    assert result.escalate is False
+
+
+def test_emotional_keyword_substring_match_is_intentionally_broad():
+    # 키워드 포함 매칭이라 오탐이 있을 수 있다 -- tech-design.md §3의 "애매하면
+    # 항상 에스컬레이션 쪽으로 fail-safe" 원칙과 일치하는 의도된 동작.
+    result = check("나 미드 정주행하다가 화나는 장면 나와서 잠깐 멈췄어")
+    assert result.escalate is True
+    assert result.reason == "감정적으로 무거운 주제"

ai-service/tests/test_generation.py

@@ -0,0 +1,75 @@
+"""Promotes generate_draft.py's manual verification into pytest -- roadmap.md
+Phase 1 §2.5. The Gemini call is mocked; this only asserts the three status
+branches (escalate/no_key/ok) and the exact request shape sent to the model,
+not real generation quality (that's blind_eval.py's job)."""
+import sys
+import types
+from unittest.mock import MagicMock
+
+from app.generation import build_user_prompt, draft_reply, last_incoming_text
+
+
+def test_last_incoming_text_strips_speaker_prefix():
+    assert last_incoming_text(["나: ㅇㅇ", "상대: 오늘 뭐해?"]) == "오늘 뭐해?"
+
+
+def test_last_incoming_text_empty_context():
+    assert last_incoming_text([]) == ""
+
+
+def test_build_user_prompt_includes_examples_and_context():
+    prompt = build_user_prompt(["ㅋㅋ 그러네"], ["상대: 안녕"])
+    assert "ㅋㅋ 그러네" in prompt
+    assert "상대: 안녕" in prompt
+
+
+def test_draft_reply_escalates_before_any_model_call():
+    # api_key is present but escalation must short-circuit before genai is
+    # even imported -- if this regresses, google.genai.Client below would
+    # need to exist/be reachable and this test would start hitting real
+    # import/network paths instead of returning early.
+    status, text = draft_reply(["ㅇㅇ"], ["상대: 계좌번호 좀 알려줘"], api_key="fake-key")
+    assert status == "escalate"
+    assert text == "금전"
+
+
+def test_draft_reply_no_key_returns_prompt_without_calling_model(monkeypatch):
+    monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+    status, text = draft_reply(["ㅇㅇ"], ["상대: 오늘 저녁 뭐 먹을래?"], api_key=None)
+    assert status == "no_key"
+    assert "오늘 저녁 뭐 먹을래" in text
+
+
+def test_draft_reply_ok_calls_gemini_with_expected_args(monkeypatch):
+    # generation.py does `from google import genai` / `from google.genai import
+    # types` *inside* draft_reply, so we stub those modules in sys.modules
+    # before the call -- this also sidesteps google-genai's real dependency
+    # chain (google-auth -> cryptography), which isn't needed for a unit test
+    # and doesn't import cleanly in every sandbox.
+    fake_response = MagicMock()
+    fake_response.text = "  ㅇㅇ 좋지  "
+    fake_client = MagicMock()
+    fake_client.models.generate_content.return_value = fake_response
+
+    fake_genai = types.ModuleType("google.genai")
+    fake_genai.Client = MagicMock(return_value=fake_client)
+    fake_types = types.ModuleType("google.genai.types")
+    fake_types.GenerateContentConfig = MagicMock(side_effect=lambda **kw: kw)
+    fake_google = types.ModuleType("google")
+    fake_google.genai = fake_genai
+
+    monkeypatch.setitem(sys.modules, "google", fake_google)
+    monkeypatch.setitem(sys.modules, "google.genai", fake_genai)
+    monkeypatch.setitem(sys.modules, "google.genai.types", fake_types)
+
+    status, text = draft_reply(
+        ["ㅇㅇ 좋지"], ["상대: 오늘 저녁 뭐 먹을래?"], model="gemini-2.5-flash", api_key="fake-key"
+    )
+
+    assert status == "ok"
+    assert text == "ㅇㅇ 좋지"  # stripped
+    fake_genai.Client.assert_called_once_with(api_key="fake-key")
+    fake_client.models.generate_content.assert_called_once()
+    _, kwargs = fake_client.models.generate_content.call_args
+    assert kwargs["model"] == "gemini-2.5-flash"
+    assert "오늘 저녁 뭐 먹을래" in kwargs["contents"]

ai-service/tests/test_main.py

@@ -0,0 +1,92 @@
+"""Promotes the ad-hoc TestClient checks used to verify /draft and
+/escalate/check into pytest -- roadmap.md Phase 1 §2.5."""
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+client = TestClient(app)
+
+
+def test_health():
+    resp = client.get("/health")
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok"}
+
+
+def test_escalate_check_money():
+    resp = client.post("/escalate/check", json={"text": "계좌번호 알려줄래?"})
+    assert resp.status_code == 200
+    assert resp.json() == {"escalate": True, "reason": "금전"}
+
+
+def test_escalate_check_benign():
+    resp = client.post("/escalate/check", json={"text": "오늘 저녁에 뭐 먹을래?"})
+    assert resp.status_code == 200
+    assert resp.json() == {"escalate": False, "reason": ""}
+
+
+def test_escalate_check_requires_text_field():
+    resp = client.post("/escalate/check", json={})
+    assert resp.status_code == 422
+
+
+def test_draft_with_style_examples_no_key(monkeypatch):
+    # No GEMINI_API_KEY in this test environment -- assert the deterministic
+    # no_key path rather than hitting the real Gemini API.
+    monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+    resp = client.post(
+        "/draft",
+        json={
+            "context_lines": ["상대: 오늘 저녁에 뭐 먹을래?"],
+            "style_examples": ["ㅇㅇ 좋지", "나도 궁금하네ㅋㅋ"],
+        },
+    )
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "no_key"
+
+
+def test_draft_escalates_without_calling_style_source_logic():
+    resp = client.post(
+        "/draft",
+        json={
+            "context_lines": ["상대: 계좌번호 좀 알려줘"],
+            "style_examples": ["ㅇㅇ 알겠어"],
+        },
+    )
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "escalate", "text": "금전"}
+
+
+def test_draft_with_history_uses_retrieval(monkeypatch):
+    monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+    resp = client.post(
+        "/draft",
+        json={
+            "context_lines": ["상대: 오늘 저녁에 뭐 먹을래?"],
+            "history": ["어제 저녁엔 라면 먹었어", "핀란드는 교육이 좋대"],
+            "k": 1,
+        },
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["status"] == "no_key"
+    # 검색된 스타일 예시가 프롬프트에 반영됐는지는 no_key 응답의 text(프롬프트
+    # 원문)로 확인 -- retrieve()가 실제로 호출·반영됐다는 증거.
+    assert "어제 저녁엔 라면 먹었어" in body["text"] or "핀란드는 교육이 좋대" in body["text"]
+
+
+def test_draft_rejects_both_style_sources():
+    resp = client.post(
+        "/draft",
+        json={
+            "context_lines": ["상대: 안녕"],
+            "style_examples": ["ㅇㅇ"],
+            "history": ["ㅇㅇ"],
+        },
+    )
+    assert resp.status_code == 422
+
+
+def test_draft_rejects_neither_style_source():
+    resp = client.post("/draft", json={"context_lines": ["상대: 안녕"]})
+    assert resp.status_code == 422

ai-service/tests/test_retrieve_style.py

@@ -0,0 +1,50 @@
+"""Promotes retrieve_style.py's manual verification into pytest -- roadmap.md
+Phase 1 §2.5. Covers the recency-vs-overlap bug fixed during PoC work: overlap
+must always outrank recency, recency only breaks ties (see retrieve_style.py's
+docstring)."""
+from app.retrieve_style import _jaccard, retrieve, tokenize
+
+
+def test_tokenize_extracts_hangul_and_alnum_only():
+    assert tokenize("핀란드는 교육이 좋대!!") == {"핀란드는", "교육이", "좋대"}
+
+
+def test_jaccard_empty_sets_score_zero():
+    assert _jaccard(set(), {"a"}) == 0.0
+    assert _jaccard({"a"}, set()) == 0.0
+
+
+def test_jaccard_identical_sets_score_one():
+    assert _jaccard({"a", "b"}, {"a", "b"}) == 1.0
+
+
+def test_keyword_overlap_outranks_recency():
+    # 핀란드 관련 예시가 훨씬 과거에 있어도, 방금 온 무관한 최신 메시지들보다
+    # 우선 검색돼야 한다 -- 가중합으로 점수를 매기면 recency가 이를 뒤집는
+    # 버그가 있었음 (poc 작업 중 발견/수정).
+    history = [
+        "핀란드는 교육이 진짜 잘 되어있대",
+        "오늘 저녁 뭐 먹지",
+        "나 요즘 잠을 못 자",
+        "어제 넷플릭스 뭐 봤어",
+    ]
+    result = retrieve(history, "핀란드는 교육이 좋대", k=1)
+    assert result == ["핀란드는 교육이 진짜 잘 되어있대"]
+
+
+def test_recency_breaks_ties_among_equal_overlap():
+    # Both candidates share exactly one token with the query and have the
+    # same union size, so their Jaccard overlap ties at 0.25 -- only then
+    # should recency (the later message) win.
+    query = "사과 바나나 포도"
+    older, newer = "사과 딸기", "바나나 수박"
+    assert _jaccard(tokenize(older), tokenize(query)) == _jaccard(tokenize(newer), tokenize(query))
+
+    result = retrieve([older, newer], query, k=1)
+    assert result == [newer]
+
+
+def test_k_limits_result_count():
+    history = [f"메시지 {i}" for i in range(10)]
+    result = retrieve(history, "메시지", k=3)
+    assert len(result) == 3

core-backend/README.md

@@ -30,9 +30,10 @@ go test ./... -v
 ```
 
 `main_test.go`가 가입/중복코드 거부(409)/메시지 저장/존재하지 않는 대화방(404)/WebSocket
-브로드캐스트/초안 생성 프록시/에스컬레이션 하드게이트(차단·통과·게이트 불능 시 fail-safe)/계정
+브로드캐스트/초안 생성 프록시/에스컬레이션 하드게이트(차단·통과·게이트 불능 시 fail-safe)/자율성
-삭제까지 전부 mock AI 서비스로 실제로 돌려서 확인한다 (`../backend/`의 Python TestClient 테스트와
+플로우(L0 차단·L1 승인·L2 화이트리스트 자동발송·L2 비대상 승인 필요·에스컬레이션의 레벨 무관 우선)/
-동일한 케이스 + Go에서 새로 추가된 것들).
+계정 삭제까지 전부 mock AI 서비스로 실제로 돌려서 확인한다 (`../backend/`의 Python TestClient
+테스트와 동일한 케이스 + Go에서 새로 추가된 것들).
 
 ## 지금 있는 것
 
@@ -63,10 +64,24 @@ go test ./... -v
 - 사람이 직접 보내는 메시지(`sender_mode` 기본값 `human`)는 이 게이트를 타지 않는다 — 본인이 직접
   쓴 말을 막을 이유가 없다.
 
+**2.2/2.5 자율성 엔진(L0~L2) 최소 오케스트레이션** (PRD.md §2.1/§2.2, roadmap.md §2.5 QA 작업 중
+필요해져서 구현)
+- `PATCH /users/{id}/twin-settings` — `{"autonomy_level": "L0"|"L1"|"L2"}`로 전역 자율성 레벨 변경
+  (가입 시 기본값은 L0, PRD.md §2.1)
+- 에스컬레이션 통과 후 트윈 발송이면 레벨을 확인: **L0**은 항상 차단(403, 초안만 가능), **L1**은
+  요청에 `approved: true`가 없으면 차단(403), **L2**는 `whitelist_rules`에 매칭되는 주제면
+  `approved` 없이도 즉시 발송, 매칭이 없으면 L1과 동일하게 승인 필요
+- 화이트리스트 매칭(`whitelistMatches`)은 v1 최소 구현 — 유저의 모든 `WhitelistRule.TopicKeyword`를
+  메시지 텍스트에 부분 문자열로 매칭. `WhitelistRule.ContactID`(상대별 화이트리스트)는 아직 무시함 —
+  대화방↔연락처 연결이 아직 모델링되어 있지 않아서(클라이언트 연락처 모델이 생긴 뒤에 다시 설계 필요)
+- 레벨/화이트리스트와 무관하게 에스컬레이션이 항상 우선한다 — L2 화이트리스트 매칭 + `approved:
+  true`여도 에스컬레이션 대상이면 무조건 차단 (테스트로 확인함)
+
 ## 아직 없는 것 (다음 워크스트림)
 
-- 자율성 엔진(L0~L2) 오케스트레이션 전체 흐름 (`roadmap.md` §2.2) — 지금은 발송 시점 하드게이트만
+- 화이트리스트 규칙 CRUD API (지금은 DB에 직접 넣는 걸로 테스트함 — 유저가 화이트리스트를 실제로
-  있고, 초안 생성→승인 대기→자동발송 분기의 나머지는 아직
+  등록하는 API는 아직 없음, 2.3 자율성 설정 화면과 같이 설계 필요)
+- 상대별(`ContactID`) 화이트리스트/자율성 예외 (지금은 전역 레벨만 지원)
 - 온디바이스 말투 이력 저장 + 서버 최소 전송 (클라이언트 책임)
 - 사후 알림 + 되돌리기 "UI" 흐름 (로그 자체는 쌓이지만, 사용자에게 보여주고 되돌리는 건 Flutter 쪽)
 - 데이터 흐름 대시보드, 온디바이스 암호화 (둘 다 Flutter 클라이언트 책임 — 이 저장소엔 SDK 없어

core-backend/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"net/http"
 	"strconv"
+	"strings"
 
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
@@ -22,6 +23,14 @@ type sendMessageRequest struct {
 	SenderID   uint       `json:"sender_id" binding:"required"`
 	Text       string     `json:"text" binding:"required"`
 	SenderMode SenderMode `json:"sender_mode"`
+	// Approved represents the human tapping "승인" on an L1 draft, or on an
+	// L2 draft outside the whitelist (PRD.md §2.2). Ignored for
+	// human-authored messages.
+	Approved bool `json:"approved"`
+}
+
+type updateTwinSettingsRequest struct {
+	AutonomyLevel AutonomyLevel `json:"autonomy_level" binding:"required"`
 }
 
 type draftMessageRequest struct {
@@ -88,7 +97,9 @@ func setupRouter(db *gorm.DB, relay *ConnectionManager, ai *AIServiceClient) *gi
 		// so no client or upstream path can bypass it (AGENTS.md absolute
 		// safety invariants). Human-authored messages are the human's own
 		// words and are never gated. On any doubt (AI service unreachable
-		// or erroring) we fail closed and block the send.
+		// or erroring) we fail closed and block the send. Escalation is
+		// checked before autonomy level, and applies regardless of level or
+		// whitelist match -- L2 auto-send never overrides it.
 		if req.SenderMode == SenderTwin {
 			result, err := ai.checkEscalation(req.Text)
 			if err != nil {
@@ -105,6 +116,33 @@ func setupRouter(db *gorm.DB, relay *ConnectionManager, ai *AIServiceClient) *gi
 				c.JSON(http.StatusForbidden, gin.H{"detail": "escalated", "reason": result.Reason})
 				return
 			}
+
+			// Autonomy gate (PRD.md §2.1/§2.2, tech-design.md §3): missing
+			// settings fail closed to L0, the documented default.
+			level := AutonomyL0
+			var settings TwinSettings
+			if err := db.Where("user_id = ?", req.SenderID).First(&settings).Error; err == nil {
+				level = settings.AutonomyLevel
+			}
+
+			switch level {
+			case AutonomyL0:
+				c.JSON(http.StatusForbidden, gin.H{"detail": "L0(비서 모드)에서는 분신 자동 발송이 허용되지 않습니다 -- 초안만 생성하고 사람이 직접 보내세요"})
+				return
+			case AutonomyL1:
+				if !req.Approved {
+					c.JSON(http.StatusForbidden, gin.H{"detail": "L1은 발송 전 사용자 승인이 필요합니다"})
+					return
+				}
+			case AutonomyL2:
+				if !req.Approved && !whitelistMatches(db, req.SenderID, req.Text) {
+					c.JSON(http.StatusForbidden, gin.H{"detail": "화이트리스트에 없는 주제는 L1과 동일하게 사용자 승인이 필요합니다"})
+					return
+				}
+			default:
+				c.JSON(http.StatusForbidden, gin.H{"detail": "알 수 없는 자율성 레벨이라 발송을 차단합니다"})
+				return
+			}
 		}
 
 		message := Message{
@@ -168,6 +206,36 @@ func setupRouter(db *gorm.DB, relay *ConnectionManager, ai *AIServiceClient) *gi
 		c.JSON(http.StatusOK, gin.H{"status": result.Status, "text": result.Text})
 	})
 
+	r.PATCH("/users/:id/twin-settings", func(c *gin.Context) {
+		userID, ok := parseUintParam(c, "id")
+		if !ok {
+			return
+		}
+
+		var req updateTwinSettingsRequest
+		if err := c.ShouldBindJSON(&req); err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
+			return
+		}
+		if req.AutonomyLevel != AutonomyL0 && req.AutonomyLevel != AutonomyL1 && req.AutonomyLevel != AutonomyL2 {
+			c.JSON(http.StatusBadRequest, gin.H{"detail": "autonomy_level must be one of L0, L1, L2"})
+			return
+		}
+
+		var settings TwinSettings
+		if err := db.Where("user_id = ?", userID).First(&settings).Error; err != nil {
+			c.JSON(http.StatusNotFound, gin.H{"detail": "twin settings not found for user"})
+			return
+		}
+		settings.AutonomyLevel = req.AutonomyLevel
+		if err := db.Save(&settings).Error; err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
+			return
+		}
+
+		c.JSON(http.StatusOK, gin.H{"user_id": userID, "autonomy_level": settings.AutonomyLevel})
+	})
+
 	r.DELETE("/users/:id", func(c *gin.Context) {
 		userID, ok := parseUintParam(c, "id")
 		if !ok {
@@ -247,6 +315,23 @@ func setupRouter(db *gorm.DB, relay *ConnectionManager, ai *AIServiceClient) *gi
 	return r
 }
 
+// whitelistMatches is a v1-minimal check: any of the user's WhitelistRule
+// keywords appearing as a substring of the message text counts as a match.
+// It intentionally ignores WhitelistRule.ContactID (per-counterpart
+// whitelisting) because conversations aren't yet linked to a Contact row --
+// that link needs its own design pass once the client's contact model
+// exists, so this only supports the "any counterpart" case for now.
+func whitelistMatches(db *gorm.DB, userID uint, text string) bool {
+	var rules []WhitelistRule
+	db.Where("user_id = ?", userID).Find(&rules)
+	for _, rule := range rules {
+		if strings.Contains(text, rule.TopicKeyword) {
+			return true
+		}
+	}
+	return false
+}
+
 func parseUintParam(c *gin.Context, name string) (uint, bool) {
 	id, err := strconv.ParseUint(c.Param(name), 10, 64)
 	if err != nil {

core-backend/main_test.go

@@ -88,6 +88,20 @@ func postJSON(t *testing.T, url string, body interface{}) *http.Response {
 	return resp
 }
 
+func setAutonomyLevel(t *testing.T, serverURL string, userID uint, level AutonomyLevel) {
+	t.Helper()
+	body, _ := json.Marshal(updateTwinSettingsRequest{AutonomyLevel: level})
+	req, _ := http.NewRequest(http.MethodPatch, serverURL+"/users/"+strconv.FormatUint(uint64(userID), 10)+"/twin-settings", bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("set autonomy level: %v", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("expected 200 setting autonomy level, got %d", resp.StatusCode)
+	}
+}
+
 func TestHealth(t *testing.T) {
 	server, _ := setupTestServer(t)
 	resp, err := http.Get(server.URL + "/health")
@@ -139,6 +153,8 @@ func TestSendMessageAndWebSocketBroadcast(t *testing.T) {
 		t.Fatalf("create conversation: %v", err)
 	}
 
+	setAutonomyLevel(t, server.URL, senderID, AutonomyL1)
+
 	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws/conversations/" +
 		strconv.FormatUint(uint64(conv.ID), 10)
 	ws, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
@@ -151,6 +167,7 @@ func TestSendMessageAndWebSocketBroadcast(t *testing.T) {
 		SenderID:   senderID,
 		Text:       "안녕하세요",
 		SenderMode: SenderTwin,
+		Approved:   true,
 	})
 	if sendResp.StatusCode != http.StatusOK {
 		t.Fatalf("expected 200 sending message, got %d", sendResp.StatusCode)
@@ -213,7 +230,11 @@ func TestTwinMessageEscalatedIsBlockedAndNotBroadcast(t *testing.T) {
 	}
 }
 
-func TestTwinMessageNotEscalatedIsSent(t *testing.T) {
+// The four tests below cover PRD.md §2.1/§2.2's L0->L1->L2 autonomy flow:
+// L0 (기본값) never auto-sends, L1 requires explicit approval, L2 auto-sends
+// only for whitelisted topics and otherwise behaves like L1.
+
+func TestTwinSendBlockedAtDefaultL0(t *testing.T) {
 	server, db := setupTestServer(t)
 
 	signupResp := postJSON(t, server.URL+"/auth/signup", signupRequest{InviteCode: "twin2", DisplayName: "하늘"})
@@ -226,19 +247,143 @@ func TestTwinMessageNotEscalatedIsSent(t *testing.T) {
 		t.Fatalf("create conversation: %v", err)
 	}
 
+	// New users default to L0 (PRD.md §2.1) -- a non-escalating twin send
+	// must still be blocked, since L0 means no auto-send at all.
 	resp := postJSON(t, server.URL+"/conversations/"+strconv.FormatUint(uint64(conv.ID), 10)+"/messages", sendMessageRequest{
 		SenderID:   senderID,
 		Text:       "ㅇㅇ 알겠어",
 		SenderMode: SenderTwin,
 	})
-	if resp.StatusCode != http.StatusOK {
+	if resp.StatusCode != http.StatusForbidden {
-		t.Fatalf("expected 200 for non-escalated twin send, got %d", resp.StatusCode)
+		t.Fatalf("expected 403 blocking twin auto-send at L0, got %d", resp.StatusCode)
+	}
+
+	var count int64
+	db.Model(&Message{}).Where("conversation_id = ?", conv.ID).Count(&count)
+	if count != 0 {
+		t.Fatalf("L0 must never persist a twin auto-send, found %d rows", count)
+	}
+}
+
+func TestTwinSendRequiresApprovalAtL1(t *testing.T) {
+	server, db := setupTestServer(t)
+
+	signupResp := postJSON(t, server.URL+"/auth/signup", signupRequest{InviteCode: "twin-l1", DisplayName: "서준"})
+	var user map[string]interface{}
+	json.NewDecoder(signupResp.Body).Decode(&user)
+	senderID := uint(user["id"].(float64))
+	setAutonomyLevel(t, server.URL, senderID, AutonomyL1)
+
+	conv := Conversation{IsGroup: false}
+	if err := db.Create(&conv).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+	convPath := server.URL + "/conversations/" + strconv.FormatUint(uint64(conv.ID), 10) + "/messages"
+
+	unapproved := postJSON(t, convPath, sendMessageRequest{SenderID: senderID, Text: "ㅇㅇ 알겠어", SenderMode: SenderTwin})
+	if unapproved.StatusCode != http.StatusForbidden {
+		t.Fatalf("expected 403 without approval at L1, got %d", unapproved.StatusCode)
+	}
+
+	approved := postJSON(t, convPath, sendMessageRequest{SenderID: senderID, Text: "ㅇㅇ 알겠어", SenderMode: SenderTwin, Approved: true})
+	if approved.StatusCode != http.StatusOK {
+		t.Fatalf("expected 200 with approval at L1, got %d", approved.StatusCode)
 	}
 
 	var count int64
 	db.Model(&Message{}).Where("conversation_id = ?", conv.ID).Count(&count)
 	if count != 1 {
-		t.Fatalf("expected 1 persisted message, got %d", count)
+		t.Fatalf("expected exactly 1 persisted message (the approved one), got %d", count)
+	}
+}
+
+func TestTwinSendAutoSendsAtL2WithWhitelistMatch(t *testing.T) {
+	server, db := setupTestServer(t)
+
+	signupResp := postJSON(t, server.URL+"/auth/signup", signupRequest{InviteCode: "twin-l2-wl", DisplayName: "가은"})
+	var user map[string]interface{}
+	json.NewDecoder(signupResp.Body).Decode(&user)
+	senderID := uint(user["id"].(float64))
+	setAutonomyLevel(t, server.URL, senderID, AutonomyL2)
+	db.Create(&WhitelistRule{UserID: senderID, TopicKeyword: "저녁"})
+
+	conv := Conversation{IsGroup: false}
+	if err := db.Create(&conv).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+
+	// No approved:true -- L2 + whitelist match should auto-send without it.
+	resp := postJSON(t, server.URL+"/conversations/"+strconv.FormatUint(uint64(conv.ID), 10)+"/messages", sendMessageRequest{
+		SenderID: senderID, Text: "저녁 뭐 먹었어?", SenderMode: SenderTwin,
+	})
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("expected 200 auto-sending whitelisted topic at L2, got %d", resp.StatusCode)
+	}
+
+	var count int64
+	db.Model(&Message{}).Where("conversation_id = ?", conv.ID).Count(&count)
+	if count != 1 {
+		t.Fatalf("expected 1 auto-sent message, got %d", count)
+	}
+}
+
+func TestTwinSendRequiresApprovalAtL2WithoutWhitelistMatch(t *testing.T) {
+	server, db := setupTestServer(t)
+
+	signupResp := postJSON(t, server.URL+"/auth/signup", signupRequest{InviteCode: "twin-l2-nowl", DisplayName: "도윤"})
+	var user map[string]interface{}
+	json.NewDecoder(signupResp.Body).Decode(&user)
+	senderID := uint(user["id"].(float64))
+	setAutonomyLevel(t, server.URL, senderID, AutonomyL2)
+	db.Create(&WhitelistRule{UserID: senderID, TopicKeyword: "저녁"})
+
+	conv := Conversation{IsGroup: false}
+	if err := db.Create(&conv).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+	convPath := server.URL + "/conversations/" + strconv.FormatUint(uint64(conv.ID), 10) + "/messages"
+
+	// Text doesn't match any whitelist keyword -- L2 falls back to L1
+	// behavior (approval required), it does not just allow or just block.
+	unapproved := postJSON(t, convPath, sendMessageRequest{SenderID: senderID, Text: "주말에 영화 볼래?", SenderMode: SenderTwin})
+	if unapproved.StatusCode != http.StatusForbidden {
+		t.Fatalf("expected 403 for non-whitelisted topic at L2 without approval, got %d", unapproved.StatusCode)
+	}
+
+	approved := postJSON(t, convPath, sendMessageRequest{SenderID: senderID, Text: "주말에 영화 볼래?", SenderMode: SenderTwin, Approved: true})
+	if approved.StatusCode != http.StatusOK {
+		t.Fatalf("expected 200 for non-whitelisted topic at L2 with approval, got %d", approved.StatusCode)
+	}
+}
+
+func TestEscalationOverridesAutonomyLevelAndWhitelist(t *testing.T) {
+	server, db := setupTestServer(t)
+
+	signupResp := postJSON(t, server.URL+"/auth/signup", signupRequest{InviteCode: "twin-l2-esc", DisplayName: "은서"})
+	var user map[string]interface{}
+	json.NewDecoder(signupResp.Body).Decode(&user)
+	senderID := uint(user["id"].(float64))
+	setAutonomyLevel(t, server.URL, senderID, AutonomyL2)
+	// Whitelisted keyword happens to be the same word that also triggers
+	// the money escalation pattern in the mock AI service.
+	db.Create(&WhitelistRule{UserID: senderID, TopicKeyword: "계좌"})
+
+	conv := Conversation{IsGroup: false}
+	if err := db.Create(&conv).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+
+	resp := postJSON(t, server.URL+"/conversations/"+strconv.FormatUint(uint64(conv.ID), 10)+"/messages", sendMessageRequest{
+		SenderID: senderID, Text: "계좌번호 알려줄게", SenderMode: SenderTwin, Approved: true,
+	})
+	if resp.StatusCode != http.StatusForbidden {
+		t.Fatalf("expected 403: escalation must override L2 whitelist match and approved:true, got %d", resp.StatusCode)
+	}
+
+	var count int64
+	db.Model(&Message{}).Where("conversation_id = ?", conv.ID).Count(&count)
+	if count != 0 {
+		t.Fatalf("escalated message must never be sent regardless of level/whitelist/approval, found %d rows", count)
 	}
 }
 

docs/roadmap.md

@@ -49,9 +49,14 @@
 - [x] Go 코어가 이 서비스를 실제로 호출하는 클라이언트 코드 (`core-backend/`에서 `AI_SERVICE_URL` 사용)
   — `core-backend/aiservice.go`(`AIServiceClient.requestDraft`) + `POST /conversations/:id/draft`
   라우트, mock AI 서비스로 정상 프록시·404·400(스타일 소스 없음) 전부 실제 테스트로 확인함
-- [ ] 자율성 엔진(L0~L2) 오케스트레이션: 에스컬레이션 게이트 → 검색 → 초안 생성 → 승인/자동발송 분기
+- [x] 자율성 엔진(L0~L2) 오케스트레이션 최소 버전 — 2.5 QA에서 "자율성 플로우 통합 테스트"를 쓰려면
-  (`tech-design.md` §3 흐름 그대로) — 이 오케스트레이션이 Go 코어와 Python AI 서비스 중 어디
+  실제 분기 로직이 있어야 해서 그때 구현함. 결정: 에스컬레이션 하드게이트(Go 코어, 이미 구현) →
-  책임인지는 구현 시작 시 정할 것 (에스컬레이션 하드게이트는 Go 코어에 두는 게 안전선 원칙상 더 맞을 수 있음)
+  레벨 확인은 Go 코어 책임(`PATCH /users/:id/twin-settings`로 레벨 변경, 메시지 저장 시 레벨별 분기).
+  L0 항상 차단, L1은 `approved:true` 필요, L2는 화이트리스트 매칭 시 즉시 자동발송·매칭 없으면 L1과
+  동일. 에스컬레이션은 레벨/화이트리스트 무관 항상 우선. **간소화한 부분**: 검색(retrieve)→초안
+  생성은 이미 있는 `/draft` 흐름을 그대로 쓰면 되므로 새로 만들지 않았고, 화이트리스트는
+  `ContactID`(상대별) 무시하고 전역 키워드 매칭만 지원 — 대화방↔연락처 연결 모델이 아직 없어서
+  (Flutter 클라이언트의 연락처 모델이 생긴 뒤 다시 설계 필요, `core-backend/README.md` 참고)
 - [ ] 온디바이스 말투 이력 저장 + 서버 최소 전송 원칙 구현
 - [ ] 사후 알림 + 되돌리기 로그 스키마/API
 
@@ -77,9 +82,16 @@
   없음)에서는 진행 불가 — 2.3과 함께 로컬 환경 대기
 
 **2.5 QA/테스트**
-- [ ] `escalation_filter.py`의 자체 테스트를 정식 테스트 스위트로 승격, `generate_draft`·`retrieve_style`도 동일하게
+- [x] `escalation_filter.py`의 자체 테스트를 정식 테스트 스위트로 승격, `generate_draft`·`retrieve_style`도
-- [ ] 자율성 플로우(L0→L1→L2) 통합 테스트
+  동일하게 — `ai-service/tests/`(pytest, 34개), SELFTEST_CASES 승격 + Gemini 호출 mock + `/health`·
-- [ ] 온보딩·채팅·설정 수동 QA
+  `/escalate/check`·`/draft` FastAPI 엔드포인트 테스트까지 포함. `poc/tone-corpus/`의 ad-hoc
+  `--selftest`는 실험 도구로 그대로 두고(승격 대상은 "실제 서비스"인 `ai-service/`), 별개로 유지
+- [x] 자율성 플로우(L0→L1→L2) 통합 테스트 — `core-backend/main_test.go`. 위 2.2 최소 오케스트레이션
+  구현과 함께: L0 차단, L1 미승인 차단/승인 시 발송, L2 화이트리스트 매칭 자동발송/비매칭 시 승인
+  필요, 에스컬레이션이 레벨·화이트리스트·승인 여부와 무관하게 항상 우선한다는 것까지 6개 케이스
+  전부 실제 테스트로 확인함
+- [ ] 온보딩·채팅·설정 수동 QA — Flutter 클라이언트가 없어 이 환경에서는 불가, 2.3과 함께 로컬
+  환경 대기
 
 **2.6 베타 배포 준비**
 - [ ] 초대 기반 베타 가입 플로우
@@ -108,14 +120,15 @@
    병행하려던 2.3 Flutter 채팅 UI 뼈대는 **이 작업 환경에 Flutter/Dart SDK가 없어 빌드 검증이
    불가능**해서 보류 — Flutter는 Windows에 SDK가 설치된 환경(본인 로컬)에서 시작
 3. [x] 2.2 AI 서비스 — `ai-service/`(Python) 완료. Go 코어→AI 서비스 연동(`core-backend/aiservice.go`,
-   `POST /conversations/:id/draft`)도 완료·테스트 통과. 남은 건 자율성 엔진 오케스트레이션, 온디바이스
+   `POST /conversations/:id/draft`)도 완료·테스트 통과. 자율성 엔진 오케스트레이션도 최소 버전으로
-   말투 이력 저장, 사후 알림/되돌리기 로그 — 이건 2.4/2.5와 겹치므로 그쪽에서 이어감. 2.3 Flutter는
+   완료(아래 5번과 함께 구현). 남은 건 온디바이스 말투 이력 저장, 사후 알림/되돌리기 로그 — 클라이언트
-   여전히 로컬 환경 대기 중
+   의존이라 2.3(Flutter) 대기
 4. [~] 2.3 나머지 UX(온보딩·설정·뱃지) — Flutter SDK가 없는 이 환경에서는 여전히 착수 불가.
    대신 2.4(안전장치 통합) 중 서버 쪽 몫(에스컬레이션 하드게이트 우회 차단, 삭제 API)을 먼저 진행함
-5. [~] 2.4/2.5 안전장치·QA — 2.4의 서버 쪽 절반(하드게이트 우회 차단, 삭제 플로우) 완료. 나머지는
+5. [x] 2.4/2.5 안전장치·QA (서버 쪽) — 2.4: 하드게이트 우회 차단, 삭제 플로우. 2.5: `ai-service`
-   클라이언트 쪽(온디바이스 암호화·데이터 흐름 대시보드, Flutter 대기)과 2.5 QA(테스트 스위트 승격,
+   pytest 스위트 승격, 자율성 플로우(L0→L1→L2) 통합 테스트 — 이걸 쓰려면 실제 분기 로직이 필요해서
-   자율성 플로우 통합 테스트) — 다음 작업
+   2.2의 자율성 오케스트레이션도 이때 같이 구현함. 남은 건 전부 클라이언트 쪽(온디바이스 암호화·
+   데이터 흐름 대시보드·수동 QA) — Flutter 대기
 6. [ ] §3 확정 (PoC 결과 필요 — 1~5번 전부 끝난 뒤에만) → 2.6 베타 오픈
 
 1~5는 PoC 실제 실행(A트랙)과 병행 가능 — PoC가 늦어져도 인프라 작업은 막히지 않는다. 다만