diff --git a/apps/api/app/routes/kb.py b/apps/api/app/routes/kb.py index cb4386e..c30ba67 100644 --- a/apps/api/app/routes/kb.py +++ b/apps/api/app/routes/kb.py @@ -356,7 +356,7 @@ async def sync_live_coach_source_packs( evaluator RAG 검색에도 올린다. source row를 먼저 upsert한 뒤 content_hash 기반 증분 색인을 수행한다. 임베딩 모델 미가용 시 BM25-only degraded 색인으로 이어진다. """ - source_rows, index_payloads = source_pack_sync.build_repo_source_pack_manifest() + source_rows, index_payloads = source_pack_sync.build_repo_source_pack_manifest(refresh=True) if not source_rows or not index_payloads: raise HTTPException( status.HTTP_422_UNPROCESSABLE_ENTITY, @@ -365,7 +365,7 @@ async def sync_live_coach_source_packs( try: async with acquire() as conn: - result = await source_pack_sync.sync_repo_source_packs(conn, apply=True) + result = await source_pack_sync.sync_repo_source_packs(conn, apply=True, refresh=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: diff --git a/apps/api/app/services/live_coach.py b/apps/api/app/services/live_coach.py index b038307..5827131 100644 --- a/apps/api/app/services/live_coach.py +++ b/apps/api/app/services/live_coach.py @@ -18,11 +18,12 @@ import hashlib from functools import lru_cache from typing import TYPE_CHECKING, Any, Literal, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from ..config import settings from ..engine_client import EngineClient, EngineError, EngineMessage, GenerateRequest, GenerateResponse from ..paths import repo_root, repo_path +from ..session_read_model import StageLabel, stage_label_or_none from . import guardrail if TYPE_CHECKING: @@ -77,12 +78,17 @@ class LiveCoachEvent(BaseModel): event_id: str session_id: str turn_seq: int - stage: str + stage: StageLabel | None = None created_at: str learner_text_excerpt: Optional[str] = None client_reply_excerpt: Optional[str] = None suggestion: LiveCoachSuggestion + @field_validator("stage", mode="before") + @classmethod + def _normalize_stage(cls, value: object) -> StageLabel | None: + return stage_label_or_none(value) + class LiveCoachInput(BaseModel): """라이브 코칭 입력. raw text는 서비스 내부에서 마스킹 후 프롬프트에 쓴다.""" @@ -126,6 +132,8 @@ _ALLOWED_KB_KINDS = { "ko_context", "microskill", } +LOCAL_SOURCE_PACK_CACHE_KEY = "repo:data/kb/live_coaching_workbook_0615.json+data/kb/live_coaching_sources/*.json" +LOCAL_SOURCE_PACK_CACHE_LIFETIME = "api-process" def _configured_model(value: str | None) -> str | None: @@ -135,6 +143,13 @@ def _configured_model(value: str | None) -> str | None: @lru_cache(maxsize=1) def _local_source_entries() -> tuple[tuple[str, dict[str, Any]], ...]: + """Load repo-managed live-coach source packs. + + Cache ownership: live turn generation keeps this process-local snapshot to + avoid per-turn file IO. Admin/CLI source-pack sync is the invalidation + boundary because that path explicitly compares repo files with kb.document. + """ + entries: list[tuple[str, dict[str, Any]]] = [] paths = [_WORKBOOK_PATH] if _SOURCE_DIR.exists(): @@ -164,6 +179,12 @@ def _local_source_entries() -> tuple[tuple[str, dict[str, Any]], ...]: ) +def clear_local_source_pack_cache() -> None: + """Invalidate the process-local source pack snapshot for admin/CLI refresh.""" + + _local_source_entries.cache_clear() + + def _local_source_payloads() -> list[dict[str, Any]]: return [payload for _, payload in _local_source_entries()] @@ -673,5 +694,6 @@ __all__ = [ "LiveCoachInput", "LiveCoachSource", "LiveCoachSuggestion", + "clear_local_source_pack_cache", "generate_live_coaching", ] diff --git a/apps/api/app/services/source_pack_sync.py b/apps/api/app/services/source_pack_sync.py index 86c348f..eb28276 100644 --- a/apps/api/app/services/source_pack_sync.py +++ b/apps/api/app/services/source_pack_sync.py @@ -40,9 +40,11 @@ class RepoSourcePackSyncResult: items: list[RepoSourcePackSyncItem] -def build_repo_source_pack_manifest() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: +def build_repo_source_pack_manifest(*, refresh: bool = False) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Return repo-managed source rows and RAG index payloads without DB access.""" + if refresh: + live_coach.clear_local_source_pack_cache() return live_coach.build_rag_source_rows(), live_coach.build_rag_index_payloads() @@ -92,7 +94,12 @@ async def _upsert_source_rows(conn: Any, source_rows: list[dict[str, Any]]) -> N ) -async def sync_repo_source_packs(conn: Any, *, apply: bool = False) -> RepoSourcePackSyncResult: +async def sync_repo_source_packs( + conn: Any, + *, + apply: bool = False, + refresh: 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 @@ -100,7 +107,7 @@ async def sync_repo_source_packs(conn: Any, *, apply: bool = False) -> RepoSourc happen only when apply=True. """ - source_rows, index_payloads = build_repo_source_pack_manifest() + source_rows, index_payloads = build_repo_source_pack_manifest(refresh=refresh) source_row_by_id = {row["source_id"]: row for row in source_rows} if apply: await _upsert_source_rows(conn, source_rows) diff --git a/apps/api/app/test_live_coach_sources.py b/apps/api/app/test_live_coach_sources.py index 01d1c67..6a069a2 100644 --- a/apps/api/app/test_live_coach_sources.py +++ b/apps/api/app/test_live_coach_sources.py @@ -2,7 +2,10 @@ from __future__ import annotations +import json +import tempfile import unittest +from pathlib import Path from typing import Any from unittest.mock import AsyncMock, patch @@ -24,6 +27,59 @@ class _SourcePackConn: class LiveCoachSourcePackTest(unittest.TestCase): + def test_manifest_refresh_reads_changed_repo_source_pack_after_cache_warm(self) -> None: + with tempfile.TemporaryDirectory() as temp_root: + root = Path(temp_root) + workbook_path = root / "live_coaching_workbook_0615.json" + source_dir = root / "live_coaching_sources" + source_dir.mkdir() + + def write_workbook(summary: str) -> None: + workbook_path.write_text( + json.dumps( + { + "source": { + "source_id": "source-a", + "title": "Source A", + "kb_kind": "theory", + "license_class": "A", + "citation": "source-a", + "external_llm_ok": True, + }, + "chunks": [ + { + "id": "chunk-a", + "summary": summary, + "citation": "source-a", + "visible_to": ["evaluator"], + } + ], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + + with ( + patch.object(live_coach, "_REPO_ROOT", root), + patch.object(live_coach, "_WORKBOOK_PATH", workbook_path), + patch.object(live_coach, "_SOURCE_DIR", source_dir), + ): + live_coach.clear_local_source_pack_cache() + write_workbook("first source pack summary") + _, first_payloads = source_pack_sync.build_repo_source_pack_manifest(refresh=True) + first_hash = first_payloads[0]["content_hash"] + + write_workbook("changed source pack summary") + _, stale_payloads = source_pack_sync.build_repo_source_pack_manifest() + self.assertEqual(stale_payloads[0]["content_hash"], first_hash) + + _, refreshed_payloads = source_pack_sync.build_repo_source_pack_manifest(refresh=True) + self.assertNotEqual(refreshed_payloads[0]["content_hash"], first_hash) + self.assertEqual(refreshed_payloads[0]["chunks"][0]["chunk_text"], "changed source pack summary") + + live_coach.clear_local_source_pack_cache() + def test_source_packs_build_rag_rows_and_index_payloads(self) -> None: rows = live_coach.build_rag_source_rows() payloads = live_coach.build_rag_index_payloads() @@ -85,6 +141,91 @@ class RagIndexPolicyTest(unittest.IsolatedAsyncioTestCase): class RepoSourcePackSyncTest(unittest.IsolatedAsyncioTestCase): + async def test_sync_refresh_after_file_change_reports_next_document_version(self) -> None: + with tempfile.TemporaryDirectory() as temp_root: + root = Path(temp_root) + workbook_path = root / "live_coaching_workbook_0615.json" + source_dir = root / "live_coaching_sources" + source_dir.mkdir() + + def write_workbook(summary: str) -> None: + workbook_path.write_text( + json.dumps( + { + "source": { + "source_id": "source-a", + "title": "Source A", + "kb_kind": "theory", + "license_class": "A", + "citation": "source-a", + "external_llm_ok": True, + }, + "chunks": [ + { + "id": "chunk-a", + "summary": summary, + "citation": "source-a", + "visible_to": ["evaluator"], + } + ], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + + with ( + patch.object(live_coach, "_REPO_ROOT", root), + patch.object(live_coach, "_WORKBOOK_PATH", workbook_path), + patch.object(live_coach, "_SOURCE_DIR", source_dir), + ): + live_coach.clear_local_source_pack_cache() + write_workbook("first source pack summary") + _, first_payloads = source_pack_sync.build_repo_source_pack_manifest(refresh=True) + first_payload = first_payloads[0] + + write_workbook("changed source pack summary") + conn = _SourcePackConn( + { + ("source-a", first_payload["doc_uri"]): { + "doc_id": 10, + "version": 2, + "content_hash": first_payload["content_hash"], + } + } + ) + result = await source_pack_sync.sync_repo_source_packs(conn, apply=False, refresh=True) + + self.assertFalse(result.applied) + self.assertEqual(result.items[0].previous_version, 2) + self.assertEqual(result.items[0].new_version, 3) + self.assertFalse(result.items[0].skipped_unchanged) + self.assertNotEqual(result.items[0].content_hash, first_payload["content_hash"]) + + live_coach.clear_local_source_pack_cache() + + def test_manifest_refresh_invalidates_live_coach_source_cache(self) -> None: + with ( + patch.object(source_pack_sync.live_coach, "clear_local_source_pack_cache") as clear_cache, + patch.object(source_pack_sync.live_coach, "build_rag_source_rows", return_value=[]) as build_rows, + patch.object(source_pack_sync.live_coach, "build_rag_index_payloads", return_value=[]) as build_payloads, + ): + source_pack_sync.build_repo_source_pack_manifest(refresh=True) + + clear_cache.assert_called_once_with() + build_rows.assert_called_once_with() + build_payloads.assert_called_once_with() + + def test_manifest_default_reuses_live_coach_source_cache(self) -> None: + with ( + patch.object(source_pack_sync.live_coach, "clear_local_source_pack_cache") as clear_cache, + patch.object(source_pack_sync.live_coach, "build_rag_source_rows", return_value=[]), + patch.object(source_pack_sync.live_coach, "build_rag_index_payloads", return_value=[]), + ): + source_pack_sync.build_repo_source_pack_manifest() + + clear_cache.assert_not_called() + def _patch_manifest(self) -> Any: row = { "source_id": "source-a", diff --git a/scripts/sync-persona-sources.py b/scripts/sync-persona-sources.py index 3348b06..17be339 100644 --- a/scripts/sync-persona-sources.py +++ b/scripts/sync-persona-sources.py @@ -59,7 +59,11 @@ 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)) + result = await source_pack_sync.sync_repo_source_packs( + conn, + apply=bool(args.apply), + refresh=True, + ) finally: await close_pool() if args.json: