perf(rag): 사전 계산 query embedding 기반 KB fan-out 병렬화

.github/workflows/test-knowledge-runtime-postgres.yml

@@ -24,9 +24,14 @@ on:
       - "apps/shared/tests/domain/test_knowledge_runtime_candidates.py"
       - "apps/shared/tests/services/test_knowledge_permission_runtime_bulk.py"
       - "apps/workflow_engine/adapters/knowledge_runtime_candidates.py"
+      - "apps/workflow_engine/adapters/rag_retrieval_connection_acquirer.py"
+      - "apps/workflow_engine/adapters/rag_retrieval_executor.py"
+      - "apps/workflow_engine/adapters/rag_retrieval_session.py"
+      - "apps/workflow_engine/application/rag_retrieval_fanout.py"
       - "apps/workflow_engine/application/runtime_retrieval/**"
       - "apps/workflow_engine/composition/runtime_retrieval.py"
       - "apps/workflow_engine/tests/adapters/test_postgres_knowledge_runtime_candidate_adapter.py"
+      - "apps/workflow_engine/tests/adapters/test_rag_retrieval_session_postgres.py"
       - ".github/workflows/test-knowledge-runtime-postgres.yml"
 
 permissions:
@@ -95,4 +100,5 @@ jobs:
           apps/workflow_engine/.venv/bin/python -m pytest
           apps/shared/tests/db/test_knowledge_runtime_snapshot_disposable_postgres.py
           apps/workflow_engine/tests/adapters/test_postgres_knowledge_runtime_candidate_adapter.py
+          apps/workflow_engine/tests/adapters/test_rag_retrieval_session_postgres.py
           -q

apps/gateway/tests/architecture/test_workflow_node_import_boundary.py

@@ -0,0 +1,83 @@
+"""Gateway와 Workflow worker 패키지의 import 경계를 검증한다."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[4]
+
+
+def test_llm_entity_import_does_not_load_worker_runtime() -> None:
+    """Data-only schema import must not require worker-only dependencies."""
+
+    script = """
+import sys
+
+from apps.workflow_engine.workflow.nodes.llm.entities import LLMNodeData
+
+assert LLMNodeData.__name__ == "LLMNodeData"
+assert "apps.workflow_engine.workflow.nodes.llm.llm_node" not in sys.modules
+"""
+    environment = os.environ.copy()
+    existing_pythonpath = environment.get("PYTHONPATH")
+    environment["PYTHONPATH"] = os.pathsep.join(
+        value
+        for value in (str(REPOSITORY_ROOT), existing_pythonpath)
+        if value
+    )
+
+    completed = subprocess.run(
+        [sys.executable, "-c", script],
+        cwd=REPOSITORY_ROOT,
+        env=environment,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+
+
+def test_node_factory_import_does_not_require_gevent_runtime() -> None:
+    """Gateway-side graph validation must not require worker-only gevent."""
+
+    script = """
+import sys
+
+
+class BlockGeventImport:
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname == "gevent" or fullname.startswith("gevent."):
+            raise ModuleNotFoundError("blocked worker-only gevent dependency")
+        return None
+
+
+sys.meta_path.insert(0, BlockGeventImport())
+
+from apps.workflow_engine.workflow.core.workflow_node_factory import NodeFactory
+
+assert "llmNode" in NodeFactory.NODE_REGISTRY
+assert "gevent" not in sys.modules
+"""
+    environment = os.environ.copy()
+    existing_pythonpath = environment.get("PYTHONPATH")
+    environment["PYTHONPATH"] = os.pathsep.join(
+        value
+        for value in (str(REPOSITORY_ROOT), existing_pythonpath)
+        if value
+    )
+
+    completed = subprocess.run(
+        [sys.executable, "-c", script],
+        cwd=REPOSITORY_ROOT,
+        env=environment,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert completed.returncode == 0, completed.stderr

apps/shared/services/tracing/metadata.py

@@ -246,9 +246,11 @@
         "total_tokens",
     },
     "rag": {
+        "candidate_resolution_latency_ms",
         "citation_ids",
         "context_token_estimate",
         "document_ids",
+        "evidence_policy_latency_ms",
         "evidence_sufficient",
         "fanout_concurrency",
         "fanout_timeout_seconds",
@@ -262,10 +264,12 @@
         "permission_filter_applied",
         "latency_ms",
         "partial_result",
+        "query_embedding_latency_ms",
         "query_rewrite_applied",
         "query_rewrite_strategy",
         "raw_content_returned",
         "rag_mode",
+        "retrieval_fanout_latency_ms",
         "retrieval_payload_id",
         "retrieval_strategy",
         "retrieved_chunk_summary_truncated",
@@ -275,6 +279,7 @@
         "score_summary",
         "selected_kb_count",
         "selected_kb_count_bucket",
+        "slowest_search_latency_ms",
         "source_tier_policy",
         "source_tier_used",
         "stored_result_count",
@@ -303,6 +308,14 @@
         "latency_ms",
     },
 }
+RAG_STAGE_LATENCY_FIELDS = {
+    "candidate_resolution_latency_ms",
+    "query_embedding_latency_ms",
+    "retrieval_fanout_latency_ms",
+    "slowest_search_latency_ms",
+    "evidence_policy_latency_ms",
+}
+MAX_RAG_STAGE_LATENCY_MS = 300_000
 RAG_RESULT_FIELDS = {
     "document_id",
     "chunk_id",
@@ -780,6 +793,14 @@ def _sanitize_rag_section(cls, value: Any) -> dict[str, Any]:
         sanitized: dict[str, Any] = {}
         for key in SPAN_SECTION_FIELDS["rag"]:
             if key in safe_value:
+                if key in RAG_STAGE_LATENCY_FIELDS:
+                    latency = safe_value[key]
+                    if (
+                        type(latency) is int
+                        and 0 <= latency <= MAX_RAG_STAGE_LATENCY_MS
+                    ):
+                        sanitized[key] = latency
+                    continue
                 sanitized_value = cls._sanitize_allowed_value(safe_value[key])
                 if sanitized_value is not None:
                     sanitized[key] = sanitized_value

apps/shared/tests/services/test_tracing_metadata.py

@@ -582,6 +582,43 @@ def test_rag_span_metadata_preserves_evidence_summary_fields_only():
     assert "raw_rewritten_query" not in metadata["rag"]
 
 
+def test_rag_span_metadata_allows_only_bounded_integer_stage_latencies():
+    metadata = TraceMetadataSanitizer.sanitize_span_metadata(
+        "llmNode",
+        {
+            "rag": {
+                "candidate_resolution_latency_ms": 0,
+                "query_embedding_latency_ms": 12,
+                "retrieval_fanout_latency_ms": 34,
+                "slowest_search_latency_ms": 56,
+                "evidence_policy_latency_ms": 300_000,
+                "per_kb_latency_ms": {"hidden-resource": 99},
+            }
+        },
+    )
+
+    assert metadata["rag"] == {
+        "candidate_resolution_latency_ms": 0,
+        "query_embedding_latency_ms": 12,
+        "retrieval_fanout_latency_ms": 34,
+        "slowest_search_latency_ms": 56,
+        "evidence_policy_latency_ms": 300_000,
+    }
+
+
+@pytest.mark.parametrize(
+    "invalid_value",
+    [True, -1, 1.5, "12", float("nan"), float("inf"), 300_001],
+)
+def test_rag_span_metadata_drops_invalid_stage_latency(invalid_value):
+    metadata = TraceMetadataSanitizer.sanitize_span_metadata(
+        "llmNode",
+        {"rag": {"retrieval_fanout_latency_ms": invalid_value}},
+    )
+
+    assert metadata.get("rag", {}) == {}
+
+
 def test_trace_detail_metadata_view_hides_error_message_and_sanitizes_metadata():
     run = SimpleNamespace(
         id=uuid.uuid4(),