From e8e08935ed65e93f170ebb917f02a7e996de6872 Mon Sep 17 00:00:00 2001 From: Yun Chan Date: Sun, 28 Jun 2026 20:12:35 +0900 Subject: [PATCH] =?UTF-8?q?=ED=8E=98=EB=A5=B4=EC=86=8C=EB=82=98=20?= =?UTF-8?q?=EC=86=8C=EC=8A=A4=ED=8C=A9=20=EB=8F=99=EA=B8=B0=ED=99=94=20?= =?UTF-8?q?=EC=A0=95=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/persona_read_model.py | 222 +++++++++++++ apps/api/app/routes/kb.py | 68 ++-- apps/api/app/routes/personas.py | 373 ++++++++-------------- apps/api/app/services/rag.py | 157 +++++++++ apps/api/app/services/source_pack_sync.py | 194 +++++++++++ apps/api/app/test_live_coach_sources.py | 112 ++++++- apps/api/app/test_persona_review.py | 64 +++- infra/db/init/03_kb.sql | 20 ++ scripts/materialize-persona-seeds.py | 108 +++++++ scripts/sync-persona-sources.py | 91 ++++++ 10 files changed, 1126 insertions(+), 283 deletions(-) create mode 100644 apps/api/app/persona_read_model.py create mode 100644 apps/api/app/services/source_pack_sync.py create mode 100644 scripts/materialize-persona-seeds.py create mode 100644 scripts/sync-persona-sources.py diff --git a/apps/api/app/persona_read_model.py b/apps/api/app/persona_read_model.py new file mode 100644 index 0000000..3038b88 --- /dev/null +++ b/apps/api/app/persona_read_model.py @@ -0,0 +1,222 @@ +"""Browser-facing persona DTOs and deterministic mappers.""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, Field + +from .persona_repository import CatalogPersona, PersonaDraftRecord, PersonaReviewAction, PersonaReviewItem +from .services.persona import PersonaCard + +JSON_OBJECT_FIELD = {"additionalProperties": True} +PersonaSourceKind = Literal["client_record", "textbook_guide", "mixed_notes"] + + +class PersonaSummary(BaseModel): + persona_id: str | None = None + code: str + version: int | None = None + status: Literal["approved"] = "approved" + display_name: str + difficulty: str + theory_target: list[str] + demographics: dict[str, Any] + presenting_summary: str + voice_preset: str | None = None + source: str = "database" + degraded: bool = False + + +class PersonaReviewSummary(BaseModel): + persona_id: str + code: str + version: int + status: Literal["draft", "review", "approved", "archived"] + display_name: str + difficulty: str + theory_target: list[str] + source_provenance: str + is_synthetic: bool + created_at: str | None = None + approved_at: str | None = None + + +class PersonaReviewDecisionRequest(BaseModel): + action: PersonaReviewAction + + +class PersonaRevisionRequest(BaseModel): + submit_for_review: bool = False + + +class PersonaDraftPayload(BaseModel): + code: str = Field(min_length=1, max_length=24) + display_name: str = Field(min_length=1, max_length=80) + difficulty: Literal["easy", "moderate", "hard"] + theory_target: list[str] = Field(default_factory=list) + demographics: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + presenting: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + history: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + big5: dict[str, float] = Field(default_factory=dict) + resistance: dict[str, float] = Field(default_factory=dict) + speech_style: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + affect_baseline: dict[str, float] = Field(default_factory=dict) + ccd: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + dsm5_dimensional: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + triggers: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) + source_provenance: str = Field(default="", max_length=240) + is_synthetic: bool = True + submit_for_review: bool = False + + +class PersonaSourceDocumentRequest(BaseModel): + filename: str = Field(min_length=1, max_length=240) + source_kind: PersonaSourceKind = "mixed_notes" + text: str = Field(min_length=20, max_length=120000) + title: str | None = Field(default=None, max_length=160) + source_note: str = Field(default="", max_length=800) + + +class PersonaSourceDocumentResponse(BaseModel): + source_id: str + doc_id: int | None + doc_uri: str + title: str + source_kind: PersonaSourceKind + kb_kind: str + license_class: Literal["A", "B", "C", "D"] = "B" + external_llm_ok: bool = True + content_hash: str + chunk_count: int + chunks_indexed: int + embedded: bool + degraded: bool = False + pii_entities_masked: list[str] = Field(default_factory=list) + + +class PersonaGenerationEvidence(BaseModel): + chunk_id: int + source_id: str + score: float + kb_kind: str + heading_path: str | None = None + excerpt: str + + +class PersonaDraftGenerateRequest(BaseModel): + source_text: str | None = Field(default=None, min_length=20, max_length=30000) + source_ids: list[str] = Field(default_factory=list, max_length=12) + source_kind: PersonaSourceKind = "mixed_notes" + code_hint: str | None = Field(default=None, max_length=24) + display_name_hint: str | None = Field(default=None, max_length=80) + difficulty: Literal["easy", "moderate", "hard"] = "moderate" + theory_target: list[str] = Field(default_factory=lambda: ["humanistic"]) + generation_goal: str = Field(default="", max_length=800) + + +class PersonaDraftGenerateResponse(BaseModel): + draft: PersonaDraftPayload + source_summary: str = "" + warnings: list[str] = Field(default_factory=list) + pii_entities_masked: list[str] = Field(default_factory=list) + source_references: list[PersonaSourceDocumentResponse] = Field(default_factory=list) + evidence_chunks: list[PersonaGenerationEvidence] = Field(default_factory=list) + + +class PersonaDraftDetail(PersonaReviewSummary): + demographics: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + presenting: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + history: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + big5: dict[str, float] + resistance: dict[str, float] + speech_style: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + affect_baseline: dict[str, float] + ccd: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + dsm5_dimensional: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + triggers: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) + + +def _first_text_value(data: dict[str, Any]) -> str: + for value in data.values(): + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +def persona_summary(entry: CatalogPersona) -> PersonaSummary: + card = entry.card + return PersonaSummary( + persona_id=entry.persona_id, + code=card.code, + version=entry.version, + status="approved", + display_name=card.display_name, + difficulty=card.difficulty, + theory_target=card.theory_target, + demographics=card.demographics, + presenting_summary=_first_text_value(card.presenting), + source=entry.source, + degraded=entry.degraded, + ) + + +def persona_review_summary(entry: PersonaReviewItem) -> PersonaReviewSummary: + return PersonaReviewSummary( + persona_id=entry.persona_id, + code=entry.code, + version=entry.version, + status=entry.status, + display_name=entry.display_name, + difficulty=entry.difficulty, + theory_target=entry.theory_target, + source_provenance=entry.source_provenance, + is_synthetic=entry.is_synthetic, + created_at=entry.created_at, + approved_at=entry.approved_at, + ) + + +def persona_draft_detail(entry: PersonaDraftRecord) -> PersonaDraftDetail: + card = entry.card + return PersonaDraftDetail( + **persona_review_summary(entry.review).model_dump(), + demographics=card.demographics, + presenting=card.presenting, + history=card.history, + big5=card.big5, + resistance=card.resistance, + speech_style=card.speech_style, + affect_baseline=card.affect_baseline, + ccd=card.ccd, + dsm5_dimensional=card.dsm5_dimensional, + triggers=card.triggers, + ) + + +def persona_card_from_draft_payload(request: PersonaDraftPayload) -> PersonaCard: + code = request.code.strip().upper() + display_name = request.display_name.strip() + if not code: + raise ValueError("persona code is required") + if not display_name: + raise ValueError("display_name is required") + theory_target = [value.strip().lower() for value in request.theory_target if value.strip()] + return PersonaCard( + code=code, + display_name=display_name, + difficulty=request.difficulty, + theory_target=theory_target, + demographics=request.demographics, + presenting=request.presenting, + history=request.history, + big5=request.big5, + resistance=request.resistance, + speech_style=request.speech_style, + affect_baseline=request.affect_baseline, + ccd=request.ccd, + dsm5_dimensional=request.dsm5_dimensional, + triggers=request.triggers, + source_provenance=request.source_provenance.strip(), + is_synthetic=request.is_synthetic, + ) diff --git a/apps/api/app/routes/kb.py b/apps/api/app/routes/kb.py index ff9adfe..cb4386e 100644 --- a/apps/api/app/routes/kb.py +++ b/apps/api/app/routes/kb.py @@ -22,7 +22,7 @@ from pydantic import BaseModel, Field from ..db import acquire, get_pool from ..deps import AIView, Principal, Role, require_role -from ..services import live_coach, rag +from ..services import rag, source_pack_sync router = APIRouter(prefix="/kb", tags=["kb"]) @@ -326,6 +326,8 @@ async def index_document( # 관리자 인덱싱은 RLS 미적용(쓰기 — kb 스키마 직접). role 주입 없이 acquire. async with acquire() as conn: result = await rag.index_document(conn, req) + except rag.IndexPolicyViolation as e: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)) from e except rag.NotConfigured as e: raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"RAG not configured: {e}") except RuntimeError as e: @@ -354,64 +356,36 @@ async def sync_live_coach_source_packs( evaluator RAG 검색에도 올린다. source row를 먼저 upsert한 뒤 content_hash 기반 증분 색인을 수행한다. 임베딩 모델 미가용 시 BM25-only degraded 색인으로 이어진다. """ - source_rows = live_coach.build_rag_source_rows() - index_payloads = live_coach.build_rag_index_payloads() + source_rows, index_payloads = source_pack_sync.build_repo_source_pack_manifest() if not source_rows or not index_payloads: raise HTTPException( status.HTTP_422_UNPROCESSABLE_ENTITY, detail="live coach source packs are empty", ) - items: list[LiveCoachSourcePackSyncItem] = [] - source_row_by_id = {row["source_id"]: row for row in source_rows} try: async with acquire() as conn: - for row in source_rows: - await conn.execute( - """ - INSERT INTO kb.source - (source_id, title, kb_kind, license_class, origin_path, citation, external_llm_ok) - VALUES ($1, $2, $3, $4, $5, $6, $7) - ON CONFLICT (source_id) DO UPDATE SET - title = EXCLUDED.title, - kb_kind = EXCLUDED.kb_kind, - license_class = EXCLUDED.license_class, - origin_path = EXCLUDED.origin_path, - citation = EXCLUDED.citation, - external_llm_ok = EXCLUDED.external_llm_ok - """, - row["source_id"], - row["title"], - row["kb_kind"], - row["license_class"], - row["origin_path"], - row["citation"], - row["external_llm_ok"], - ) - for payload in index_payloads: - if payload["source_id"] not in source_row_by_id: - continue - result = await rag.index_document(conn, rag.IndexRequest(**payload)) - items.append( - LiveCoachSourcePackSyncItem( - source_id=payload["source_id"], - doc_id=result.doc_id, - chunks_indexed=result.chunks_indexed, - skipped_unchanged=result.skipped_unchanged, - embedded=result.embedded, - degraded=result.degraded, - ) - ) + result = await source_pack_sync.sync_repo_source_packs(conn, apply=True) except rag.NotConfigured as exc: raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"RAG not configured: {exc}") from exc except RuntimeError as exc: raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"DB not ready: {exc}") from exc return LiveCoachSourcePackSyncResponse( - sources_upserted=len(source_rows), - chunks_indexed=sum(item.chunks_indexed for item in items), - skipped_unchanged=sum(1 for item in items if item.skipped_unchanged), - embedded=all(item.embedded for item in items) if items else False, - degraded=any(item.degraded for item in items), - items=items, + sources_upserted=result.sources_upserted, + chunks_indexed=result.chunks_indexed, + skipped_unchanged=result.skipped_unchanged, + embedded=result.embedded, + degraded=result.degraded, + items=[ + LiveCoachSourcePackSyncItem( + source_id=item.source_id, + doc_id=item.doc_id, + chunks_indexed=item.chunks_indexed, + skipped_unchanged=item.skipped_unchanged, + embedded=item.embedded, + degraded=item.degraded, + ) + for item in result.items + ], ) diff --git a/apps/api/app/routes/personas.py b/apps/api/app/routes/personas.py index 903baf7..b3765f9 100644 --- a/apps/api/app/routes/personas.py +++ b/apps/api/app/routes/personas.py @@ -9,16 +9,11 @@ import uuid from typing import Annotated, Any, Literal from fastapi import APIRouter, Depends, HTTPException, Response, status -from pydantic import BaseModel, Field from ..db import acquire from ..deps import CurrentPrincipal, Principal, Role, require_role from ..deps import AIView from ..persona_repository import ( - CatalogPersona, - PersonaDraftRecord, - PersonaReviewAction, - PersonaReviewItem, archive_persona_family, create_persona_draft, create_persona_revision_from_existing, @@ -28,15 +23,30 @@ from ..persona_repository import ( update_persona_draft, update_persona_review_status, ) +from ..persona_read_model import ( + PersonaDraftDetail, + PersonaDraftGenerateRequest, + PersonaDraftGenerateResponse, + PersonaDraftPayload, + PersonaGenerationEvidence, + PersonaReviewDecisionRequest, + PersonaReviewSummary, + PersonaRevisionRequest, + PersonaSourceDocumentRequest, + PersonaSourceDocumentResponse, + PersonaSourceKind, + PersonaSummary, + persona_card_from_draft_payload, + persona_draft_detail, + persona_review_summary, + persona_summary, +) from ..engine_client import EngineError, EngineMessage, GenerateRequest, engine_client from ..services import rag from ..services.guardrail import mask_pii -from ..services.persona import PersonaCard router = APIRouter(prefix="/personas", tags=["personas"]) TeacherOrAdmin = Annotated[Principal, Depends(require_role(Role.TEACHER, Role.ADMIN))] -JSON_OBJECT_FIELD = {"additionalProperties": True} -PersonaSourceKind = Literal["client_record", "textbook_guide", "mixed_notes"] PERSONA_SOURCE_KB_KIND: dict[str, str] = { "client_record": "diagnostic", "textbook_guide": "theory", @@ -47,215 +57,45 @@ PERSONA_SOURCE_CITATION: dict[str, str] = { "textbook_guide": "교수자 첨부 교재/가이드 환언·발췌 근거 — 저작권 검수 필요", "mixed_notes": "교수자 첨부 혼합 메모 PII 마스킹 파생본", } +PERSONA_DRAFT_PROMPT_BUNDLE_ID = "persona-draft-rag" +PERSONA_DRAFT_PROMPT_BUNDLE_VERSION = "2026-06-28.1" +PERSONA_DRAFT_SYSTEM_PROMPT = ( + "출력은 반드시 structured_schema를 따른다. code는 P숫자 형식을 선호하되 " + "힌트가 없으면 빈 문자열 대신 임시값 P로 둔다. source_provenance에는 " + "RAG source_id와 첨부 근거 기반 초안임을 남긴다. evidence chunk id를 " + "임상 필드 본문에 그대로 노출하지 않는다." +) +PERSONA_DRAFT_USER_PROMPT_PREAMBLE = ( + "너는 Vignette 임상 콘텐츠 저작 보조자다. 아래 RAG 근거 청크만 바탕으로 교육용 " + "가상내담자 페르소나 초안을 만든다. 첨부 원문은 KB 문서가 SSOT이며, 근거 밖 내용을 " + "임의로 꾸며 핵심 임상 정보처럼 쓰지 않는다. 실제 개인정보는 이미 마스킹됐으며, " + "원문 표현을 복사하지 말고 " + "범주화·합성화된 임상 훈련용 설정으로 변환한다. CCD/DSM/역린은 런타임 내부 설정이므로 " + "내담자 발화에 직접 노출되지 않는 형태로 작성한다." +) -class PersonaSummary(BaseModel): - persona_id: str | None = None - code: str - version: int | None = None - status: Literal["approved"] = "approved" - display_name: str - difficulty: str - theory_target: list[str] - demographics: dict[str, Any] - presenting_summary: str - voice_preset: str | None = None - source: str = "database" - degraded: bool = False - - -class PersonaReviewSummary(BaseModel): - persona_id: str - code: str - version: int - status: Literal["draft", "review", "approved", "archived"] - display_name: str - difficulty: str - theory_target: list[str] - source_provenance: str - is_synthetic: bool - created_at: str | None = None - approved_at: str | None = None - - -class PersonaReviewDecisionRequest(BaseModel): - action: PersonaReviewAction - - -class PersonaRevisionRequest(BaseModel): - submit_for_review: bool = False - - -class PersonaDraftPayload(BaseModel): - code: str = Field(min_length=1, max_length=24) - display_name: str = Field(min_length=1, max_length=80) - difficulty: Literal["easy", "moderate", "hard"] - theory_target: list[str] = Field(default_factory=list) - demographics: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - presenting: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - history: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - big5: dict[str, float] = Field(default_factory=dict) - resistance: dict[str, float] = Field(default_factory=dict) - speech_style: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - affect_baseline: dict[str, float] = Field(default_factory=dict) - ccd: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - dsm5_dimensional: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - triggers: dict[str, Any] = Field(default_factory=dict, json_schema_extra=JSON_OBJECT_FIELD) - source_provenance: str = Field(default="", max_length=240) - is_synthetic: bool = True - submit_for_review: bool = False - - -class PersonaSourceDocumentRequest(BaseModel): - filename: str = Field(min_length=1, max_length=240) - source_kind: PersonaSourceKind = "mixed_notes" - text: str = Field(min_length=20, max_length=120000) - title: str | None = Field(default=None, max_length=160) - source_note: str = Field(default="", max_length=800) - - -class PersonaSourceDocumentResponse(BaseModel): - source_id: str - doc_id: int | None - doc_uri: str - title: str - source_kind: PersonaSourceKind - kb_kind: str - license_class: Literal["A", "B", "C", "D"] = "B" - external_llm_ok: bool = True - content_hash: str - chunk_count: int - chunks_indexed: int - embedded: bool - degraded: bool = False - pii_entities_masked: list[str] = Field(default_factory=list) - - -class PersonaGenerationEvidence(BaseModel): - chunk_id: int - source_id: str - score: float - kb_kind: str - heading_path: str | None = None - excerpt: str - - -class PersonaDraftGenerateRequest(BaseModel): - source_text: str | None = Field(default=None, min_length=20, max_length=30000) - source_ids: list[str] = Field(default_factory=list, max_length=12) - source_kind: PersonaSourceKind = "mixed_notes" - code_hint: str | None = Field(default=None, max_length=24) - display_name_hint: str | None = Field(default=None, max_length=80) - difficulty: Literal["easy", "moderate", "hard"] = "moderate" - theory_target: list[str] = Field(default_factory=lambda: ["humanistic"]) - generation_goal: str = Field(default="", max_length=800) - - -class PersonaDraftGenerateResponse(BaseModel): - draft: PersonaDraftPayload - source_summary: str = "" - warnings: list[str] = Field(default_factory=list) - pii_entities_masked: list[str] = Field(default_factory=list) - source_references: list[PersonaSourceDocumentResponse] = Field(default_factory=list) - evidence_chunks: list[PersonaGenerationEvidence] = Field(default_factory=list) - - -class PersonaDraftDetail(PersonaReviewSummary): - demographics: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - presenting: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - history: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - big5: dict[str, float] - resistance: dict[str, float] - speech_style: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - affect_baseline: dict[str, float] - ccd: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - dsm5_dimensional: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - triggers: dict[str, Any] = Field(json_schema_extra=JSON_OBJECT_FIELD) - - -def _first_text_value(data: dict[str, Any]) -> str: - for value in data.values(): - if isinstance(value, str) and value.strip(): - return value.strip() - return "" - - -def _summary(entry: CatalogPersona) -> PersonaSummary: - card = entry.card - return PersonaSummary( - persona_id=entry.persona_id, - code=card.code, - version=entry.version, - status="approved", - display_name=card.display_name, - difficulty=card.difficulty, - theory_target=card.theory_target, - demographics=card.demographics, - presenting_summary=_first_text_value(card.presenting), - source=entry.source, - degraded=entry.degraded, +def _persona_draft_prompt_bundle() -> dict[str, str]: + payload = "\n".join( + [ + PERSONA_DRAFT_PROMPT_BUNDLE_ID, + PERSONA_DRAFT_PROMPT_BUNDLE_VERSION, + PERSONA_DRAFT_SYSTEM_PROMPT, + PERSONA_DRAFT_USER_PROMPT_PREAMBLE, + ] ) + return { + "id": PERSONA_DRAFT_PROMPT_BUNDLE_ID, + "version": PERSONA_DRAFT_PROMPT_BUNDLE_VERSION, + "hash": hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12], + } -def _review_summary(entry: PersonaReviewItem) -> PersonaReviewSummary: - return PersonaReviewSummary( - persona_id=entry.persona_id, - code=entry.code, - version=entry.version, - status=entry.status, - display_name=entry.display_name, - difficulty=entry.difficulty, - theory_target=entry.theory_target, - source_provenance=entry.source_provenance, - is_synthetic=entry.is_synthetic, - created_at=entry.created_at, - approved_at=entry.approved_at, - ) - - -def _draft_detail(entry: PersonaDraftRecord) -> PersonaDraftDetail: - card = entry.card - return PersonaDraftDetail( - **_review_summary(entry.review).model_dump(), - demographics=card.demographics, - presenting=card.presenting, - history=card.history, - big5=card.big5, - resistance=card.resistance, - speech_style=card.speech_style, - affect_baseline=card.affect_baseline, - ccd=card.ccd, - dsm5_dimensional=card.dsm5_dimensional, - triggers=card.triggers, - ) - - -def _card_from_draft_payload(request: PersonaDraftPayload) -> PersonaCard: - code = request.code.strip().upper() - display_name = request.display_name.strip() - if not code: - raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail="persona code is required") - if not display_name: - raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail="display_name is required") - theory_target = [value.strip().lower() for value in request.theory_target if value.strip()] - return PersonaCard( - code=code, - display_name=display_name, - difficulty=request.difficulty, - theory_target=theory_target, - demographics=request.demographics, - presenting=request.presenting, - history=request.history, - big5=request.big5, - resistance=request.resistance, - speech_style=request.speech_style, - affect_baseline=request.affect_baseline, - ccd=request.ccd, - dsm5_dimensional=request.dsm5_dimensional, - triggers=request.triggers, - source_provenance=request.source_provenance.strip(), - is_synthetic=request.is_synthetic, - ) +def _card_from_draft_payload(request: PersonaDraftPayload): + try: + return persona_card_from_draft_payload(request) + except ValueError as exc: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc def _safe_doc_segment(value: str) -> str: @@ -299,6 +139,63 @@ def _source_content_hash(chunks: list[dict[str, Any]]) -> str: return h.hexdigest() +def _raw_source_content_hash(request: PersonaSourceDocumentRequest) -> str: + h = hashlib.sha256() + h.update(request.text.encode("utf-8")) + h.update(request.filename.encode("utf-8")) + h.update(request.source_kind.encode("utf-8")) + return h.hexdigest() + + +def _raw_source_artifact_id(source_id: str) -> str: + return f"{source_id}_raw" + + +async def _record_persona_raw_source_artifact( + conn: Any, + *, + request: PersonaSourceDocumentRequest, + principal: Principal, + source_id: str, + doc_uri: str, + content_hash: str, + masked_entities: list[str], + license_class: str, +) -> None: + summary = { + "source_kind": request.source_kind, + "filename": request.filename, + "title": _source_title(request), + "pii_entities_masked": masked_entities, + "derived_source_id": source_id, + "raw_text_not_indexed": True, + "storage": "hash_only", + } + await conn.execute( + """ + INSERT INTO kb.raw_source_artifact + (raw_source_id, derived_source_id, owner_user_id, doc_uri, content_hash, + storage_uri, sealed_blob, license_class, redaction_summary, external_llm_ok) + VALUES ($1, $2, $3::uuid, $4, $5, NULL, NULL, $6, $7::jsonb, FALSE) + ON CONFLICT (raw_source_id) DO UPDATE SET + derived_source_id = EXCLUDED.derived_source_id, + owner_user_id = EXCLUDED.owner_user_id, + doc_uri = EXCLUDED.doc_uri, + content_hash = EXCLUDED.content_hash, + license_class = EXCLUDED.license_class, + redaction_summary = EXCLUDED.redaction_summary, + external_llm_ok = FALSE + """, + _raw_source_artifact_id(source_id), + source_id, + principal.user_id, + f"{doc_uri}.raw", + content_hash, + license_class, + json.dumps(summary, ensure_ascii=False), + ) + + def _source_title(request: PersonaSourceDocumentRequest) -> str: return (request.title or request.filename).strip() @@ -315,6 +212,8 @@ async def _register_persona_source_document( f"persona-authoring/{principal.user_id}/" f"{source_id}/{_safe_doc_segment(request.filename)}" ) + raw_source_id = _raw_source_artifact_id(source_id) + raw_content_hash = _raw_source_content_hash(request) kb_kind = PERSONA_SOURCE_KB_KIND.get(request.source_kind, "ko_context") license_class: Literal["A", "B", "C", "D"] = "B" external_llm_ok = True @@ -340,7 +239,10 @@ async def _register_persona_source_document( "pii_entities_masked": masked.entities, "license_class": license_class, "external_llm_ok": external_llm_ok, - "raw_source_not_stored": True, + "raw_source_id": raw_source_id, + "raw_source_content_hash": raw_content_hash, + "raw_source_not_indexed": True, + "raw_source_storage": "hash_only", }, "token_count": max(1, len(chunk) // 4), } @@ -374,7 +276,19 @@ async def _register_persona_source_document( request.filename, PERSONA_SOURCE_CITATION.get(request.source_kind, "교수자 첨부 PII 마스킹 파생본"), ) + await _record_persona_raw_source_artifact( + conn, + request=request, + principal=principal, + source_id=source_id, + doc_uri=doc_uri, + content_hash=raw_content_hash, + masked_entities=masked.entities, + license_class=license_class, + ) result = await rag.index_document(conn, index_req) + except rag.IndexPolicyViolation as exc: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc except rag.NotConfigured as exc: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, @@ -698,7 +612,7 @@ async def list_personas(response: Response, _principal: CurrentPrincipal) -> lis else: response.headers["X-Vignette-Catalog-Source"] = "database" - return [_summary(entry) for entry in personas] + return [persona_summary(entry) for entry in personas] @router.get("/review", response_model=list[PersonaReviewSummary]) @@ -712,7 +626,7 @@ async def list_persona_reviews(principal: TeacherOrAdmin) -> list[PersonaReviewS status.HTTP_503_SERVICE_UNAVAILABLE, detail="persona review queue database unavailable", ) from exc - return [_review_summary(entry) for entry in queue] + return [persona_review_summary(entry) for entry in queue] @router.post("/drafts", response_model=PersonaReviewSummary, status_code=status.HTTP_201_CREATED) @@ -736,7 +650,7 @@ async def create_persona_draft_route( status.HTTP_503_SERVICE_UNAVAILABLE, detail="persona draft database unavailable", ) from exc - return _review_summary(created) + return persona_review_summary(created) @router.post("/{persona_id}/revisions", response_model=PersonaDraftDetail, status_code=status.HTTP_201_CREATED) @@ -763,7 +677,7 @@ async def create_persona_revision_route( ) from exc if record is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="approved persona not found") - return _draft_detail(record) + return persona_draft_detail(record) @router.delete("/{persona_id}", response_model=PersonaReviewSummary) @@ -788,7 +702,7 @@ async def archive_persona_route( ) from exc if archived is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="persona not found") - return _review_summary(archived) + return persona_review_summary(archived) @router.post("/sources", response_model=PersonaSourceDocumentResponse, status_code=status.HTTP_201_CREATED) @@ -861,13 +775,9 @@ async def generate_persona_draft_route( query=evidence_query or "페르소나 저작 근거", ) evidence_text = _format_generation_evidence(evidence) + prompt_bundle = _persona_draft_prompt_bundle() prompt = ( - "너는 Vignette 임상 콘텐츠 저작 보조자다. 아래 RAG 근거 청크만 바탕으로 교육용 " - "가상내담자 페르소나 초안을 만든다. 첨부 원문은 KB 문서가 SSOT이며, 근거 밖 내용을 " - "임의로 꾸며 핵심 임상 정보처럼 쓰지 않는다. 실제 개인정보는 이미 마스킹됐으며, " - "원문 표현을 복사하지 말고 " - "범주화·합성화된 임상 훈련용 설정으로 변환한다. CCD/DSM/역린은 런타임 내부 설정이므로 " - "내담자 발화에 직접 노출되지 않는 형태로 작성한다.\n\n" + f"{PERSONA_DRAFT_USER_PROMPT_PREAMBLE}\n\n" f"자료 종류: {request.source_kind}\n" f"RAG source_ids: {source_ids}\n" f"코드 힌트: {code_hint or '미정'}\n" @@ -883,12 +793,7 @@ async def generate_persona_draft_route( messages=[ EngineMessage( role="system", - content=( - "출력은 반드시 structured_schema를 따른다. code는 P숫자 형식을 선호하되 " - "힌트가 없으면 빈 문자열 대신 임시값 P로 둔다. source_provenance에는 " - "RAG source_id와 첨부 근거 기반 초안임을 남긴다. evidence chunk id를 " - "임상 필드 본문에 그대로 노출하지 않는다." - ), + content=PERSONA_DRAFT_SYSTEM_PROMPT, ), EngineMessage(role="user", content=prompt), ], @@ -897,6 +802,7 @@ async def generate_persona_draft_route( structured_schema=_persona_generation_schema(), metadata={ "feature": "persona_draft_generation", + "prompt_bundle": prompt_bundle, "source_kind": request.source_kind, "source_ids": source_ids, }, @@ -911,6 +817,7 @@ async def generate_persona_draft_route( payload = response.structured or _json_payload_from_generation(response.text) draft = _coerce_generated_draft(payload, request) provenance = ( + f"prompt={prompt_bundle['id']}@{prompt_bundle['version']}#{prompt_bundle['hash']}; " f"RAG sources={','.join(source_ids)}; " f"chunks={','.join(str(item.chunk_id) for item in evidence)}" ) @@ -947,7 +854,7 @@ async def get_persona_draft_route( ) from exc if record is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="persona draft not found") - return _draft_detail(record) + return persona_draft_detail(record) @router.put("/drafts/{persona_id}", response_model=PersonaReviewSummary) @@ -975,7 +882,7 @@ async def update_persona_draft_route( ) from exc if updated is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="persona draft not found or locked") - return _review_summary(updated) + return persona_review_summary(updated) @router.post("/review/{persona_id}", response_model=PersonaReviewSummary) @@ -1005,4 +912,4 @@ async def decide_persona_review( status.HTTP_404_NOT_FOUND, detail="persona review item not found or not pending review", ) - return _review_summary(updated) + return persona_review_summary(updated) diff --git a/apps/api/app/services/rag.py b/apps/api/app/services/rag.py index 5ee12c5..6be7e0b 100644 --- a/apps/api/app/services/rag.py +++ b/apps/api/app/services/rag.py @@ -43,6 +43,10 @@ class NotConfigured(RuntimeError): """ +class IndexPolicyViolation(ValueError): + """Index request violates RAG source isolation policy.""" + + # ════════════════════════════════════════════════════════════════════════════ # 1. 정책 4-튜플 (설계서 §4.3) — 사전필터 + 회수가중치 + 리랭킹목표 + 주입방식 # 3-AI 가 *같은 물리 테이블 kb.chunk* 를 다른 정책으로 검색한다. @@ -149,6 +153,34 @@ class RetrievalResult: degraded: bool = False # reranker/embed fallback 여부(투명성) +@dataclass(slots=True) +class EpisodicTurnInput: + """app.turn_embedding writer input. + + Only masked, client-visible client utterances should be passed here. Raw text, + evaluator payloads, counselor turns, and CCD/answer-key material are not part of + this contract. + """ + + turn_id: str + session_id: str + case_id: str + seq: int + text_masked: str + speaker: str = "client" + visible_to: Sequence[str] = ("client", "counselor", "evaluator") + + +@dataclass(slots=True) +class EpisodicEmbeddingWriteResult: + """Best-effort turn_embedding writer result.""" + + candidates: int + inserted: int + skipped: int + degraded: bool = False + + # CRAG 게이트 임계값(설계서 §3.7 top1_score). 미달이면 호출부가 "관찰 프레이밍"으로 다운그레이드. # 주: 임의 가정값 — Phase 3 파일럿에서 분포 측정 후 확정(M14, "검증됨" 금지). CRAG_TOP1_THRESHOLD = 0.35 @@ -572,6 +604,91 @@ ORDER BY d.s_dense DESC LIMIT $4 """ +_TURN_EMBEDDING_INSERT_SQL = """ +INSERT INTO app.turn_embedding + (turn_id, case_id, session_id, seq, dense, sparse, context_prefix) +VALUES ($1::uuid, $2::uuid, $3::uuid, $4, $5::vector, $6::jsonb, $7) +ON CONFLICT (turn_id) DO NOTHING +""" + + +def episodic_turn_inputs_from_records( + *, + session_id: str, + case_id: str, + turns: Sequence[Any], +) -> list[EpisodicTurnInput]: + """Build safe episodic embedding inputs from TurnRecord-like objects. + + The writer is intentionally narrow: only client speaker turns that are visible + to the client, have a DB turn_id, and have non-empty masked text are eligible. + Raw `text`, evaluations, counselor messages, and evaluator-only turns are + ignored by construction. + """ + inputs: list[EpisodicTurnInput] = [] + for turn in turns: + turn_id = getattr(turn, "turn_id", None) + text_masked = str(getattr(turn, "text_masked", "") or "").strip() + visible_to = tuple(getattr(turn, "visible_to", ()) or ()) + if not turn_id or not text_masked: + continue + if getattr(turn, "speaker", None) != "client": + continue + if "client" not in visible_to: + continue + inputs.append( + EpisodicTurnInput( + turn_id=str(turn_id), + session_id=session_id, + case_id=case_id, + seq=int(getattr(turn, "turn_seq", 0) or 0), + text_masked=text_masked, + speaker="client", + visible_to=visible_to, + ) + ) + return inputs + + +async def write_persona_turn_embeddings( + conn: "asyncpg.Connection", + *, + turns: Sequence[EpisodicTurnInput], +) -> EpisodicEmbeddingWriteResult: + """Index masked client-visible client utterances into app.turn_embedding. + + This is a technical writer only. It does not infer relationship or clinical + facts, and it does not store raw text. Missing BGE-M3/pgvector is reported as + NotConfigured so callers can skip without breaking session persistence. + """ + candidates = len(turns) + inserted = 0 + for turn in turns: + if turn.speaker != "client" or "client" not in tuple(turn.visible_to or ()): + continue + text = turn.text_masked.strip() + if not text: + continue + eq = await asyncio.to_thread(embed_query, text) + result = await conn.execute( + _TURN_EMBEDDING_INSERT_SQL, + turn.turn_id, + turn.case_id, + turn.session_id, + int(turn.seq), + _vector_literal(eq.dense), + json.dumps(eq.sparse), + None, + ) + if isinstance(result, str) and result.endswith(" 1"): + inserted += 1 + return EpisodicEmbeddingWriteResult( + candidates=candidates, + inserted=inserted, + skipped=max(0, candidates - inserted), + degraded=False, + ) + async def retrieve_persona_memory( conn: "asyncpg.Connection", @@ -694,6 +811,40 @@ def _content_hash(chunks: list[dict[str, Any]]) -> str: return h.hexdigest() +def _truthy_meta_flag(meta: Any, *keys: str) -> bool: + if not isinstance(meta, dict): + return False + for key in keys: + if bool(meta.get(key)): + return True + return False + + +def _validate_index_chunks(req: IndexRequest) -> None: + """Fail closed before raw source artifacts can enter kb.chunk. + + `sensitivity=3` means 원천격리. Those materials must stay outside the + embedding/FTS index because kb.chunk is retrievable by design. + """ + for index, chunk in enumerate(req.chunks): + meta = chunk.get("meta") or {} + sensitivity = chunk.get("sensitivity") + try: + sensitivity_int = int(sensitivity) if sensitivity is not None else 0 + except (TypeError, ValueError): + sensitivity_int = 0 + if sensitivity_int >= 3 or _truthy_meta_flag( + meta, + "raw_source", + "raw_source_artifact", + "raw_source_isolated", + ): + raise IndexPolicyViolation( + "raw source artifacts must not be indexed in kb.chunk " + f"(source_id={req.source_id}, chunk_index={index})" + ) + + async def index_document( conn: "asyncpg.Connection", req: IndexRequest, @@ -711,6 +862,7 @@ async def index_document( Raises: NotConfigured — DB(kb 스키마/vector) 미가용. """ + _validate_index_chunks(req) content_hash = req.content_hash or _content_hash(req.chunks) # (1) 증분 — 동일 source/uri/version 활성본의 content_hash 비교 @@ -825,14 +977,19 @@ async def index_document( __all__ = [ "NotConfigured", + "IndexPolicyViolation", "AIRole", "RetrievalPolicy", "POLICIES", "RetrievedChunk", "RetrievalResult", + "EpisodicTurnInput", + "EpisodicEmbeddingWriteResult", "CRAG_TOP1_THRESHOLD", "EmbeddedQuery", "embed_query", + "episodic_turn_inputs_from_records", + "write_persona_turn_embeddings", "apply_contextual_prefix", "search_kb", "retrieve_persona_memory", diff --git a/apps/api/app/services/source_pack_sync.py b/apps/api/app/services/source_pack_sync.py new file mode 100644 index 0000000..86c348f --- /dev/null +++ b/apps/api/app/services/source_pack_sync.py @@ -0,0 +1,194 @@ +"""Repo-managed source pack sync helpers. + +This keeps CLI and admin API sync behavior on the same content_hash/version +path. The source pack content itself stays in data files owned outside this +service. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from . import live_coach, rag + + +@dataclass(slots=True) +class RepoSourcePackSyncItem: + source_id: str + doc_uri: str + previous_version: int | None + new_version: int + content_hash: str + doc_id: int | None + chunks_indexed: int + skipped_unchanged: bool + embedded: bool + degraded: bool = False + applied: bool = False + + +@dataclass(slots=True) +class RepoSourcePackSyncResult: + sources_upserted: int + manifest_count: int + chunks_indexed: int + skipped_unchanged: int + embedded: bool + degraded: bool + applied: bool + items: list[RepoSourcePackSyncItem] + + +def build_repo_source_pack_manifest() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """Return repo-managed source rows and RAG index payloads without DB access.""" + + return live_coach.build_rag_source_rows(), live_coach.build_rag_index_payloads() + + +def _latest_version(row: Any | None) -> int | None: + if not row: + return None + value = row["version"] + return int(value) if value is not None else None + + +async def _latest_active_document(conn: Any, *, source_id: str, doc_uri: str) -> Any | None: + return await conn.fetchrow( + """ + SELECT doc_id, version, content_hash + FROM kb.document + WHERE source_id = $1 AND doc_uri = $2 AND is_active + ORDER BY version DESC + LIMIT 1 + """, + source_id, + doc_uri, + ) + + +async def _upsert_source_rows(conn: Any, source_rows: list[dict[str, Any]]) -> None: + for row in source_rows: + await conn.execute( + """ + INSERT INTO kb.source + (source_id, title, kb_kind, license_class, origin_path, citation, external_llm_ok) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (source_id) DO UPDATE SET + title = EXCLUDED.title, + kb_kind = EXCLUDED.kb_kind, + license_class = EXCLUDED.license_class, + origin_path = EXCLUDED.origin_path, + citation = EXCLUDED.citation, + external_llm_ok = EXCLUDED.external_llm_ok + """, + row["source_id"], + row["title"], + row["kb_kind"], + row["license_class"], + row["origin_path"], + row["citation"], + row["external_llm_ok"], + ) + + +async def sync_repo_source_packs(conn: Any, *, apply: bool = False) -> RepoSourcePackSyncResult: + """Compare or apply repo-managed source packs against kb.document. + + Dry-run mode still reads the active DB document row so it can report whether + the next apply would skip or materialize a new document version. DB writes + happen only when apply=True. + """ + + source_rows, index_payloads = build_repo_source_pack_manifest() + source_row_by_id = {row["source_id"]: row for row in source_rows} + if apply: + await _upsert_source_rows(conn, source_rows) + + items: list[RepoSourcePackSyncItem] = [] + for payload in index_payloads: + source_id = str(payload["source_id"]) + if source_id not in source_row_by_id: + continue + doc_uri = str(payload["doc_uri"]) + content_hash = str(payload["content_hash"]) + latest = await _latest_active_document(conn, source_id=source_id, doc_uri=doc_uri) + previous_version = _latest_version(latest) + requested_version = max(1, int(payload.get("version") or 1)) + new_version = requested_version + if previous_version is not None: + new_version = max(requested_version, previous_version + 1) + + if latest and latest["content_hash"] == content_hash: + items.append( + RepoSourcePackSyncItem( + source_id=source_id, + doc_uri=doc_uri, + previous_version=previous_version, + new_version=previous_version or requested_version, + content_hash=content_hash, + doc_id=int(latest["doc_id"]), + chunks_indexed=0, + skipped_unchanged=True, + embedded=False, + degraded=False, + applied=apply, + ) + ) + continue + + if not apply: + items.append( + RepoSourcePackSyncItem( + source_id=source_id, + doc_uri=doc_uri, + previous_version=previous_version, + new_version=new_version, + content_hash=content_hash, + doc_id=None, + chunks_indexed=0, + skipped_unchanged=False, + embedded=False, + degraded=False, + applied=False, + ) + ) + continue + + index_payload = dict(payload) + index_payload["version"] = new_version + result = await rag.index_document(conn, rag.IndexRequest(**index_payload)) + items.append( + RepoSourcePackSyncItem( + source_id=source_id, + doc_uri=doc_uri, + previous_version=previous_version, + new_version=new_version, + content_hash=content_hash, + doc_id=result.doc_id, + chunks_indexed=result.chunks_indexed, + skipped_unchanged=result.skipped_unchanged, + embedded=result.embedded, + degraded=result.degraded, + applied=True, + ) + ) + + return RepoSourcePackSyncResult( + sources_upserted=len(source_rows) if apply else 0, + manifest_count=len(index_payloads), + chunks_indexed=sum(item.chunks_indexed for item in items), + skipped_unchanged=sum(1 for item in items if item.skipped_unchanged), + embedded=all(item.embedded for item in items) if items else False, + degraded=any(item.degraded for item in items), + applied=apply, + items=items, + ) + + +__all__ = [ + "RepoSourcePackSyncItem", + "RepoSourcePackSyncResult", + "build_repo_source_pack_manifest", + "sync_repo_source_packs", +] diff --git a/apps/api/app/test_live_coach_sources.py b/apps/api/app/test_live_coach_sources.py index 7140dd9..01d1c67 100644 --- a/apps/api/app/test_live_coach_sources.py +++ b/apps/api/app/test_live_coach_sources.py @@ -3,8 +3,24 @@ from __future__ import annotations import unittest +from typing import Any +from unittest.mock import AsyncMock, patch -from .services import live_coach +from .services import live_coach, rag, source_pack_sync + + +class _SourcePackConn: + def __init__(self, latest: dict[tuple[str, str], dict[str, Any]] | None = None) -> None: + self.latest = latest or {} + self.execute_calls: list[tuple[str, tuple[Any, ...]]] = [] + self.fetchrow_calls: list[tuple[str, tuple[Any, ...]]] = [] + + async def execute(self, query: str, *args: Any) -> None: + self.execute_calls.append((query, args)) + + async def fetchrow(self, query: str, *args: Any) -> dict[str, Any] | None: + self.fetchrow_calls.append((query, args)) + return self.latest.get((str(args[0]), str(args[1]))) class LiveCoachSourcePackTest(unittest.TestCase): @@ -39,5 +55,99 @@ class LiveCoachSourcePackTest(unittest.TestCase): self.assertTrue(all(chunk["sensitivity"] == 2 for chunk in dsm["chunks"])) +class RagIndexPolicyTest(unittest.IsolatedAsyncioTestCase): + async def test_index_document_rejects_raw_source_chunks_before_db(self) -> None: + class NoDbConn: + async def fetchrow(self, *_args: Any, **_kwargs: Any) -> None: + raise AssertionError("raw source validation must run before DB access") + + async def execute(self, *_args: Any, **_kwargs: Any) -> None: + raise AssertionError("raw source validation must run before DB access") + + req = rag.IndexRequest( + source_id="raw-source", + doc_uri="raw-source.txt", + chunks=[ + { + "seq": 0, + "chunk_text": "원문 축어록", + "visible_to": [], + "sensitivity": 3, + "meta": {"raw_source_artifact": True}, + } + ], + ) + + with self.assertRaises(rag.IndexPolicyViolation) as caught: + await rag.index_document(NoDbConn(), req) # type: ignore[arg-type] + + self.assertIn("raw source artifacts must not be indexed", str(caught.exception)) + + +class RepoSourcePackSyncTest(unittest.IsolatedAsyncioTestCase): + def _patch_manifest(self) -> Any: + row = { + "source_id": "source-a", + "title": "Source A", + "kb_kind": "theory", + "license_class": "A", + "origin_path": "data/source-a.json", + "citation": "source-a", + "external_llm_ok": True, + } + payload = { + "source_id": "source-a", + "doc_uri": "repo/source-a.json", + "version": 1, + "content_hash": "new-hash", + "chunks": [{"seq": 0, "chunk_text": "body", "visible_to": ["evaluator"], "sensitivity": 2}], + } + return patch.object(source_pack_sync, "build_repo_source_pack_manifest", return_value=([row], [payload])) + + async def test_dry_run_reports_next_version_without_writes(self) -> None: + conn = _SourcePackConn({("source-a", "repo/source-a.json"): {"doc_id": 10, "version": 2, "content_hash": "old-hash"}}) + + with self._patch_manifest(), patch.object(source_pack_sync.rag, "index_document", AsyncMock()) as index_document: + result = await source_pack_sync.sync_repo_source_packs(conn, apply=False) + + self.assertFalse(result.applied) + self.assertEqual(result.sources_upserted, 0) + self.assertEqual(conn.execute_calls, []) + index_document.assert_not_awaited() + self.assertEqual(result.items[0].previous_version, 2) + self.assertEqual(result.items[0].new_version, 3) + self.assertFalse(result.items[0].skipped_unchanged) + + async def test_apply_bumps_changed_document_version(self) -> None: + conn = _SourcePackConn({("source-a", "repo/source-a.json"): {"doc_id": 10, "version": 2, "content_hash": "old-hash"}}) + seen_versions: list[int] = [] + + async def fake_index_document(_: Any, request: rag.IndexRequest) -> rag.IndexResult: + seen_versions.append(request.version) + return rag.IndexResult(doc_id=11, chunks_indexed=1, skipped_unchanged=False, embedded=False, degraded=True) + + with self._patch_manifest(), patch.object(source_pack_sync.rag, "index_document", fake_index_document): + result = await source_pack_sync.sync_repo_source_packs(conn, apply=True) + + self.assertTrue(result.applied) + self.assertEqual(result.sources_upserted, 1) + self.assertEqual(len(conn.execute_calls), 1) + self.assertEqual(seen_versions, [3]) + self.assertEqual(result.items[0].doc_id, 11) + self.assertEqual(result.items[0].chunks_indexed, 1) + self.assertTrue(result.degraded) + + async def test_apply_skips_same_content_hash(self) -> None: + conn = _SourcePackConn({("source-a", "repo/source-a.json"): {"doc_id": 10, "version": 2, "content_hash": "new-hash"}}) + + with self._patch_manifest(), patch.object(source_pack_sync.rag, "index_document", AsyncMock()) as index_document: + result = await source_pack_sync.sync_repo_source_packs(conn, apply=True) + + index_document.assert_not_awaited() + self.assertEqual(result.skipped_unchanged, 1) + self.assertEqual(result.items[0].doc_id, 10) + self.assertEqual(result.items[0].new_version, 2) + + if __name__ == "__main__": unittest.main() diff --git a/apps/api/app/test_persona_review.py b/apps/api/app/test_persona_review.py index 1ec2550..3f1c100 100644 --- a/apps/api/app/test_persona_review.py +++ b/apps/api/app/test_persona_review.py @@ -2,7 +2,10 @@ from __future__ import annotations +import importlib.util +import json import unittest +from pathlib import Path from typing import Any from unittest.mock import AsyncMock, patch @@ -245,6 +248,44 @@ def _draft_payload( class PersonaApprovalBoundaryTest(unittest.IsolatedAsyncioTestCase): + def test_persona_seed_runner_reports_repo_manifest_without_db(self) -> None: + script_path = Path(__file__).resolve().parents[3] / "scripts" / "materialize-persona-seeds.py" + spec = importlib.util.spec_from_file_location("materialize_persona_seeds", script_path) + self.assertIsNotNone(spec) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + rows = module._manifest_rows() + + self.assertEqual([row["code"] for row in rows], ["P1", "P2", "P3", "P4", "P5", "P6", "P7"]) + self.assertEqual({row["version"] for row in rows}, {persona_repository.SEED_VERSION}) + self.assertEqual(rows[0]["persona_id"], str(persona_repository.seed_persona_id("P1"))) + self.assertIn("source_provenance", rows[0]) + + async def test_persona_seed_runner_apply_initializes_and_closes_pool(self) -> None: + script_path = Path(__file__).resolve().parents[3] / "scripts" / "materialize-persona-seeds.py" + spec = importlib.util.spec_from_file_location("materialize_persona_seeds_apply", script_path) + self.assertIsNotNone(spec) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + init_pool = AsyncMock() + materialize = AsyncMock(return_value=7) + close_pool = AsyncMock() + with ( + patch.object(module, "init_pool", init_pool), + patch.object(module, "materialize_seed_personas", materialize), + patch.object(module, "close_pool", close_pool), + ): + exit_code = await module._main_async(module.parse_args(["--apply"])) + + self.assertEqual(exit_code, 0) + init_pool.assert_awaited_once() + materialize.assert_awaited_once() + close_pool.assert_awaited_once() + async def test_repository_loads_repo_persona_files_for_p4_to_p7(self) -> None: cards = persona_repository.load_file_personas() @@ -753,12 +794,16 @@ class PersonaReviewQueueTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(response.draft.code, "P8") self.assertEqual(response.draft.triggers["sore_spots"], ["버림받음"]) self.assertEqual(response.source_summary, "관계 상실 후 불안이 높아진 사례") - self.assertEqual(response.pii_entities_masked, ["PHONE"]) + self.assertIn("PHONE", response.pii_entities_masked) self.assertEqual(response.source_references[0].source_id, "persona_authoring_test") self.assertEqual(response.evidence_chunks[0].chunk_id, 44) self.assertIn("RAG sources=persona_authoring_test", response.draft.source_provenance) self.assertIn("chunks=44", response.draft.source_provenance) + self.assertIn("prompt=persona-draft-rag@2026-06-28.1#", response.draft.source_provenance) self.assertEqual(len(captured), 1) + self.assertEqual(captured[0].metadata["prompt_bundle"]["id"], "persona-draft-rag") + self.assertEqual(captured[0].metadata["prompt_bundle"]["version"], "2026-06-28.1") + self.assertRegex(captured[0].metadata["prompt_bundle"]["hash"], r"^[0-9a-f]{12}$") sent_text = captured[0].messages[-1].content self.assertNotIn("010-1234-5678", sent_text) self.assertIn("[PHONE]", sent_text) @@ -799,7 +844,7 @@ class PersonaReviewQueueTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(response.kb_kind, "diagnostic") self.assertEqual(response.license_class, "B") self.assertTrue(response.external_llm_ok) - self.assertEqual(response.pii_entities_masked, ["PHONE"]) + self.assertIn("PHONE", response.pii_entities_masked) self.assertEqual(len(captured_index), 1) index_req = captured_index[0] self.assertTrue(index_req.source_id.startswith("persona_authoring_")) @@ -808,9 +853,24 @@ class PersonaReviewQueueTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(index_req.chunks[0]["visible_to"], ["evaluator"]) self.assertEqual(index_req.chunks[0]["sensitivity"], 2) self.assertEqual(index_req.chunks[0]["meta"]["source_kind"], "client_record") + self.assertTrue(index_req.chunks[0]["meta"]["raw_source_not_indexed"]) + self.assertEqual(index_req.chunks[0]["meta"]["raw_source_storage"], "hash_only") + self.assertTrue(index_req.chunks[0]["meta"]["raw_source_id"].endswith("_raw")) + self.assertRegex(index_req.chunks[0]["meta"]["raw_source_content_hash"], r"^[0-9a-f]{64}$") + self.assertNotIn("raw_source_not_stored", index_req.chunks[0]["meta"]) source_query, source_args = conn.execute_calls[0] self.assertIn("INSERT INTO kb.source", source_query) self.assertEqual(source_args[2], "diagnostic") + raw_query, raw_args = conn.execute_calls[1] + self.assertIn("INSERT INTO kb.raw_source_artifact", raw_query) + self.assertEqual(raw_args[0], index_req.chunks[0]["meta"]["raw_source_id"]) + self.assertEqual(raw_args[1], index_req.source_id) + self.assertEqual(raw_args[2], "00000000-0000-0000-0000-000000000901") + self.assertEqual(raw_args[5], "B") + raw_summary = json.loads(raw_args[6]) + self.assertEqual(raw_summary["derived_source_id"], index_req.source_id) + self.assertTrue(raw_summary["raw_text_not_indexed"]) + self.assertEqual(raw_summary["storage"], "hash_only") async def test_learner_cannot_create_persona_draft_route(self) -> None: with patch.object( diff --git a/infra/db/init/03_kb.sql b/infra/db/init/03_kb.sql index 89c3dfc..0491544 100644 --- a/infra/db/init/03_kb.sql +++ b/infra/db/init/03_kb.sql @@ -30,6 +30,26 @@ CREATE TABLE IF NOT EXISTS kb.source ( created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +-- ============================================================================= +-- 1-B. 원문 격리 아티팩트 (RAG 색인 금지) +-- raw_source_artifact 는 kb.chunk/embedding/FTS 로 들어가지 않는 별도 보관 증거다. +-- ============================================================================= +CREATE TABLE IF NOT EXISTS kb.raw_source_artifact ( + raw_source_id TEXT PRIMARY KEY, + derived_source_id TEXT REFERENCES kb.source(source_id), + owner_user_id UUID, + doc_uri TEXT NOT NULL, + content_hash TEXT NOT NULL, + storage_uri TEXT, + sealed_blob BYTEA, + license_class CHAR(1) NOT NULL CHECK (license_class IN ('A','B','C','D')), + redaction_summary JSONB NOT NULL DEFAULT '{}'::jsonb, + external_llm_ok BOOLEAN NOT NULL DEFAULT FALSE CHECK (external_llm_ok = FALSE), + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS idx_raw_source_artifact_derived + ON kb.raw_source_artifact(derived_source_id); + -- ============================================================================= -- 2. 문서 + 버전 (graphify content_hash 증분 인덱싱 차용) -- ============================================================================= diff --git a/scripts/materialize-persona-seeds.py b/scripts/materialize-persona-seeds.py new file mode 100644 index 0000000..86c92c4 --- /dev/null +++ b/scripts/materialize-persona-seeds.py @@ -0,0 +1,108 @@ +"""Materialize built-in persona seed versions on demand. + +Default mode is a dry run: report the repository persona seed manifest without +touching the database. Use --apply when an operator intentionally wants to run +the existing idempotent DB materializer. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import sys +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[1] +API_ROOT = REPO_ROOT / "apps" / "api" +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + + +from app.persona_repository import ( # noqa: E402 + SEED_VERSION, + built_in_personas, + materialize_seed_personas, + seed_persona_id, +) +from app.db import close_pool, init_pool # noqa: E402 + + +def _manifest_rows() -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for card in built_in_personas(): + rows.append( + { + "code": card.code, + "version": SEED_VERSION, + "persona_id": str(seed_persona_id(card.code)), + "display_name": card.display_name, + "difficulty": card.difficulty, + "theory_target": list(card.theory_target), + "source_provenance": card.source_provenance, + "synthetic": bool(card.is_synthetic), + } + ) + return rows + + +def _print_text(*, applied: bool, count: int | None, rows: list[dict[str, Any]]) -> None: + mode = "APPLY" if applied else "DRY-RUN" + print(f"persona seed materializer: {mode}") + print(f"seed_version: {SEED_VERSION}") + print(f"manifest_count: {len(rows)}") + if count is not None: + print(f"materialize_calls: {count}") + for row in rows: + theories = ",".join(row["theory_target"]) + print( + f"- {row['code']} v{row['version']} {row['persona_id']} " + f"{row['display_name']} [{theories}]" + ) + + +async def _main_async(args: argparse.Namespace) -> int: + rows = _manifest_rows() + count: int | None = None + if args.apply: + await init_pool() + try: + count = await materialize_seed_personas() + finally: + await close_pool() + payload = { + "applied": bool(args.apply), + "seed_version": SEED_VERSION, + "manifest_count": len(rows), + "materialize_calls": count, + "personas": rows, + } + if args.json: + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + else: + _print_text(applied=bool(args.apply), count=count, rows=rows) + return 0 + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Report or apply the repository persona seed/version manifest. " + "Dry-run is the default; --apply performs DB writes through the " + "existing idempotent materializer." + ) + ) + parser.add_argument("--apply", action="store_true", help="write missing seed versions to DB") + parser.add_argument("--json", action="store_true", help="emit machine-readable JSON") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(list(argv if argv is not None else sys.argv[1:])) + return asyncio.run(_main_async(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/sync-persona-sources.py b/scripts/sync-persona-sources.py new file mode 100644 index 0000000..3348b06 --- /dev/null +++ b/scripts/sync-persona-sources.py @@ -0,0 +1,91 @@ +"""Sync repo-managed persona/evaluator source packs into KB. + +Default mode is a DB-backed dry run: compare repo source pack content hashes +against active kb.document rows and report the versions that would be used. +Use --apply to upsert kb.source rows and index changed documents. +""" + +from __future__ import annotations + +import argparse +import asyncio +import dataclasses +import json +import sys +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[1] +API_ROOT = REPO_ROOT / "apps" / "api" +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + + +from app.db import acquire, close_pool, init_pool # noqa: E402 +from app.services import source_pack_sync # noqa: E402 + + +def _to_payload(result: source_pack_sync.RepoSourcePackSyncResult) -> dict[str, Any]: + return { + "applied": result.applied, + "sources_upserted": result.sources_upserted, + "manifest_count": result.manifest_count, + "chunks_indexed": result.chunks_indexed, + "skipped_unchanged": result.skipped_unchanged, + "embedded": result.embedded, + "degraded": result.degraded, + "items": [dataclasses.asdict(item) for item in result.items], + } + + +def _print_text(result: source_pack_sync.RepoSourcePackSyncResult) -> None: + mode = "APPLY" if result.applied else "DRY-RUN" + print(f"persona source pack sync: {mode}") + print(f"manifest_count: {result.manifest_count}") + print(f"sources_upserted: {result.sources_upserted}") + print(f"chunks_indexed: {result.chunks_indexed}") + print(f"skipped_unchanged: {result.skipped_unchanged}") + for item in result.items: + previous = "-" if item.previous_version is None else str(item.previous_version) + action = "skip" if item.skipped_unchanged else ("apply" if result.applied else "would_apply") + print( + f"- {item.source_id} {action} v{previous}->v{item.new_version} " + f"{item.content_hash[:12]} chunks={item.chunks_indexed}" + ) + + +async def _main_async(args: argparse.Namespace) -> int: + await init_pool() + try: + async with acquire(role="admin") as conn: + result = await source_pack_sync.sync_repo_source_packs(conn, apply=bool(args.apply)) + finally: + await close_pool() + if args.json: + print(json.dumps(_to_payload(result), ensure_ascii=False, indent=2, sort_keys=True)) + else: + _print_text(result) + return 0 + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Compare or apply repo-managed persona/evaluator source packs. " + "Dry-run is the default and reads DB state without writes; --apply " + "upserts source rows and indexes changed document versions." + ) + ) + parser.add_argument("--apply", action="store_true", help="write changed source packs into KB") + parser.add_argument("--json", action="store_true", help="emit machine-readable JSON") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(list(argv if argv is not None else sys.argv[1:])) + return asyncio.run(_main_async(args)) + + +if __name__ == "__main__": + raise SystemExit(main())