fix(email): chunk oversized embedding inputs

CHANGELOG.md

@@ -1,4 +1,5 @@
 ## [Unreleased]
+- 긴 이메일·첨부 본문을 의미 단위 청크로 임베딩한 뒤 기존 email/attachment 벡터 계약으로 평균화하고, 청크 요청·벡터 누적을 제한된 창으로 처리합니다. OpenAI `text-embedding-3-*`에는 저장 차원(`1536`)을 직접 요청하도록 보강했습니다. 합성 메일 fixture 5건(70청크)과 provider 요청 계약으로 1,536차원 벡터 경로를 검증했으며, 실행 시 선택한 임베딩 제공자에 본문·파싱된 첨부 텍스트를 전송할 수 있습니다. 회사 기밀 데이터는 fixture·commit·PR·log에 포함하지 않습니다.
 - EmailDetail 테스트가 지원하지 않는 스레드 병합/분리 버튼을 `textContent`뿐 아니라 `aria-label`과 `title` 접근 가능 이름으로도 검출하도록 바꿔, 아이콘 전용 버튼 회귀를 놓치지 않습니다.
 
 ### 캘린더 충돌 (Status-weighted conflicts)

backend/services/batch_embedding_service.py

@@ -32,6 +32,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 import uuid
 from dataclasses import dataclass
@@ -61,6 +62,11 @@
 # Poll budget while the orchestrator drains the batch through pg-llm-batch.
 _ORCHESTRATOR_POLL_INTERVAL_SECONDS = 1.0
 _ORCHESTRATOR_MAX_POLLS = 30
+# Keep each JSON request below the orchestrator's body budget with envelope
+# headroom. Import chunks are normally <= 1,000 characters, but the byte check
+# also protects direct callers that provide longer or multibyte inputs.
+_ORCHESTRATOR_MAX_INPUTS_PER_REQUEST = 32
+_ORCHESTRATOR_MAX_INPUT_BYTES = 48 * 1024
 _SUCCESS_STATUSES = frozenset({"completed", "succeeded"})
 
 # Cache the (im)port result so repeated imports don't re-probe sys.path.
@@ -114,6 +120,14 @@ def has_local_fallback(self) -> bool:
         return bool(self.local_dsn)
 
 
+@dataclass(frozen=True)
+class BatchEmbeddingPartial:
+    """Completed prefix plus inputs that still need the normal fallback path."""
+
+    completed_vectors: list[list[float]]
+    pending_texts: list[str]
+
+
 async def resolve_batch_embedding_settings(
     session: AsyncSession,
     *,
@@ -145,9 +159,7 @@ async def resolve_batch_embedding_settings(
         attribution_service=_clean(
             getattr(tenant_config, "batch_attribution_service", None)
         ),
-        attribution_team=_clean(
-            getattr(tenant_config, "batch_attribution_team", None)
-        ),
+        attribution_team=_clean(getattr(tenant_config, "batch_attribution_team", None)),
         attribution_group=_clean(
             getattr(tenant_config, "batch_attribution_group", None)
         ),
@@ -176,15 +188,16 @@ async def try_batch_import_embeddings(
     user_id: str,
     organization_id: str | None,
     dimension: int = STORAGE_EMBEDDING_DIMENSION,
-) -> list[list[float]] | None:
+) -> list[list[float]] | BatchEmbeddingPartial | None:
     """Route bulk embeddings through the batch path, or ``None`` to fall back.
 
     On success returns one fitted vector per input text (original order). The
     primary path submits to contextual-orchestrator; only if the orchestrator is
     unconfigured or unavailable does it consider the local ``pg-llm-batch``
-    package fallback. Any failure returns ``None`` so the caller uses its
-    per-item path. The run is recorded in ``llm_batch_jobs`` / ``llm_batch_items``
-    for observability.
+    package fallback. A later partition failure returns a completed prefix plus
+    only the unfinished inputs so the caller does not resend successful work.
+    The run is recorded in ``llm_batch_jobs`` / ``llm_batch_items`` for
+    observability.
     """
     if not texts:
         return None
@@ -198,7 +211,7 @@ async def try_batch_import_embeddings(
     model = settings.model or embedding_provider.embedding_model
 
     if settings.has_orchestrator:
-        result = await _run_orchestrator_batch(
+        result = await _run_orchestrator_batches(
             session,
             texts,
             settings=settings,
@@ -229,6 +242,121 @@ async def try_batch_import_embeddings(
 # --- Primary path: contextual-orchestrator batch API ------------------------
 
 
+def _serialized_orchestrator_payload_bytes(
+    inputs: list[str],
+    *,
+    model: str,
+    endpoint_alias: str | None,
+    metadata: dict[str, str],
+) -> int:
+    """Return the UTF-8 size of the request envelope sent to the orchestrator."""
+    payload = {
+        "model": model,
+        "endpoint": endpoint_alias,
+        "inputs": inputs,
+        "metadata": metadata,
+    }
+    return len(
+        json.dumps(payload, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
+    )
+
+
+def _partition_orchestrator_inputs(
+    texts: list[str],
+    *,
+    model: str = "",
+    endpoint_alias: str | None = None,
+    metadata: dict[str, str] | None = None,
+) -> list[list[str]] | None:
+    """Partition inputs by count and serialized JSON request bytes."""
+    request_metadata = metadata or {}
+    partitions: list[list[str]] = []
+    current: list[str] = []
+    for text in texts:
+        candidate = [*current, text]
+        if current and (
+            len(candidate) > _ORCHESTRATOR_MAX_INPUTS_PER_REQUEST
+            or _serialized_orchestrator_payload_bytes(
+                candidate,
+                model=model,
+                endpoint_alias=endpoint_alias,
+                metadata=request_metadata,
+            )
+            > _ORCHESTRATOR_MAX_INPUT_BYTES
+        ):
+            partitions.append(current)
+            candidate = [text]
+        if (
+            _serialized_orchestrator_payload_bytes(
+                candidate,
+                model=model,
+                endpoint_alias=endpoint_alias,
+                metadata=request_metadata,
+            )
+            > _ORCHESTRATOR_MAX_INPUT_BYTES
+        ):
+            return None
+        current = candidate
+    if current:
+        partitions.append(current)
+    return partitions
+
+
+async def _run_orchestrator_batches(
+    session: AsyncSession,
+    texts: list[str],
+    *,
+    settings: BatchEmbeddingSettings,
+    model: str,
+    user_id: str,
+    organization_id: str | None,
+    dimension: int,
+) -> list[list[float]] | BatchEmbeddingPartial | None:
+    """Submit bounded requests and concatenate vectors in original order."""
+    metadata = _attribution_metadata(
+        settings=settings,
+        user_id=user_id,
+        organization_id=organization_id,
+    )
+    partitions = _partition_orchestrator_inputs(
+        texts,
+        model=model,
+        endpoint_alias=settings.endpoint_alias,
+        metadata=metadata,
+    )
+    if partitions is None:
+        logger.warning(
+            "Orchestrator batch input exceeded one-request byte budget; falling back: "
+            "text_count=%s",
+            len(texts),
+        )
+        return None
+
+    vectors: list[list[float]] = []
+    for partition_index, partition in enumerate(partitions):
+        partition_vectors = await _run_orchestrator_batch(
+            session,
+            partition,
+            settings=settings,
+            model=model,
+            user_id=user_id,
+            organization_id=organization_id,
+            dimension=dimension,
+        )
+        if partition_vectors is None:
+            if not vectors:
+                return None
+            pending_texts = [
+                text for remaining in partitions[partition_index:] for text in remaining
+            ]
+            return BatchEmbeddingPartial(
+                completed_vectors=vectors,
+                pending_texts=pending_texts,
+            )
+        vectors.extend(partition_vectors)
+    return vectors
+
+
 async def _run_orchestrator_batch(
     session: AsyncSession,
     texts: list[str],
@@ -372,9 +500,7 @@ async def _submit_and_await(
     if status in _SUCCESS_STATUSES and document.get("embeddings") is not None:
         return document
     if status in ("failed", "error", "canceled"):
-        raise EmbeddingGenerationError(
-            f"orchestrator batch rejected: status={status}"
-        )
+        raise EmbeddingGenerationError(f"orchestrator batch rejected: status={status}")
 
     batch_id = document.get("batch_id") or document.get("id")
     if not batch_id:

backend/services/email_import_service.py

@@ -26,12 +26,16 @@
     KnowledgeGraphEdgeRecord,
 )
 from services.archive import extract_backup_async
-from services.batch_embedding_service import try_batch_import_embeddings
+from services.batch_embedding_service import (
+    BatchEmbeddingPartial,
+    try_batch_import_embeddings,
+)
 from services.content_graph import ParseResult, parse_content
 from services.email_dedupe_service import strong_email_fingerprint
 from services.email_parser import EmailData, parse_eml_bytes
 from services.embedding import (
     STORAGE_EMBEDDING_DIMENSION,
+    chunk_text,
     fit_embedding_vector,
     generate_embeddings,
 )
@@ -52,10 +56,14 @@
 
 EMBEDDING_DIMENSION = STORAGE_EMBEDDING_DIMENSION
 MAX_IMPORT_UPLOADS = 10
-MAX_IMPORT_UPLOAD_BYTES = 20 * 1024 * 1024
+# Transport safety ceiling only; parser and embedding chunking must accept
+# sources larger than 20 MiB without confusing the request guard for a parser
+# limit.
+MAX_IMPORT_UPLOAD_BYTES = 64 * 1024 * 1024
 MAX_IMPORT_EML_FILES = 100
 MAX_IMPORT_EMAILS_PER_OWNER = 1000
 MAX_UPLOAD_FILENAME_DECODE_ROUNDS = 8
+MAX_EMBEDDING_CHUNKS_PER_WINDOW = 32
 SUPPORTED_EMAIL_IMPORT_SUFFIXES = frozenset({".eml", ".mbox", ".zip"})
 EMAIL_IMPORT_QUOTA_LOCK_NAMESPACE = "naruon-email-import-quota"
 logger = logging.getLogger(__name__)
@@ -288,15 +296,52 @@ async def _extract_and_generate_embeddings(
     batch_context: "EmailImportBatchContext | None" = None,
 ) -> tuple[list[dict], list[list[float]]]:
     attachment_payloads = list(parsed.get("attachments", []))
-    embedding_texts = [str(parsed.get("body") or "")]
-    embedding_texts.extend(
-        str(attachment.get("content") or "") for attachment in attachment_payloads
-    )
-    fitted_embeddings = await _generate_import_embeddings(
-        embedding_texts,
-        embedding_provider=embedding_provider,
-        batch_context=batch_context,
+    body_parse_content = parsed.get("body_parse_content")
+    source_texts = [
+        str(
+            body_parse_content
+            if body_parse_content is not None
+            else parsed.get("body") or ""
+        )
+    ]
+    source_texts.extend(
+        str(
+            ""
+            if (attachment.get("parse_status") or "parsed") != "parsed"
+            else (
+                attachment.get("parse_content")
+                if attachment.get("parse_content") is not None
+                else attachment.get("content") or ""
+            )
+        )
+        for attachment in attachment_payloads
     )
+    fitted_embeddings: list[list[float]] = []
+    for source_text in source_texts:
+        source_chunks = chunk_text(source_text)
+        if not source_chunks:
+            fitted_embeddings.append(_zero_embedding())
+            continue
+
+        vector_sum: list[float] | None = None
+        vector_count = 0
+        for start in range(0, len(source_chunks), MAX_EMBEDDING_CHUNKS_PER_WINDOW):
+            chunk_embeddings = await _generate_import_embeddings(
+                source_chunks[start : start + MAX_EMBEDDING_CHUNKS_PER_WINDOW],
+                embedding_provider=embedding_provider,
+                batch_context=batch_context,
+            )
+            for embedding in chunk_embeddings:
+                if vector_sum is None:
+                    vector_sum = [0.0] * len(embedding)
+                for index, value in enumerate(embedding):
+                    vector_sum[index] += value
+                vector_count += 1
+        fitted_embeddings.append(
+            [value / vector_count for value in vector_sum]
+            if vector_sum and vector_count
+            else _zero_embedding()
+        )
     return attachment_payloads, fitted_embeddings
 
 
@@ -394,7 +439,11 @@ def _fallback_attachment_parser_key(
         return "calendar"
     if parse_content_type == "text/html":
         return "html"
-    if parse_content_type in {"text/markdown", "text/x-markdown", "application/markdown"}:
+    if parse_content_type in {
+        "text/markdown",
+        "text/x-markdown",
+        "application/markdown",
+    }:
         return "markdown"
     if parse_content_type == "text/plain":
         return "plain_text"
@@ -586,9 +635,9 @@ def add_edge(
             item.segment_path,
         ),
     ):
-        segments_by_source[
-            (segment.source_kind, segment.source_record_uid)
-        ].append(segment)
+        segments_by_source[(segment.source_kind, segment.source_record_uid)].append(
+            segment
+        )
         add_edge(
             edge_kind="node_has_segment",
             edge_path=f"{segment.content_node.node_path}/has/{segment.segment_path}",
@@ -603,8 +652,7 @@ def add_edge(
             add_edge(
                 edge_kind="segment_next",
                 edge_path=(
-                    f"{source_segment.segment_path}/next/"
-                    f"{target_segment.segment_path}"
+                    f"{source_segment.segment_path}/next/{target_segment.segment_path}"
                 ),
                 source_kind=source_segment.source_kind,
                 source_record_uid=source_segment.source_record_uid,
@@ -628,8 +676,7 @@ def add_edge(
             add_edge(
                 edge_kind="heading_contains_segment",
                 edge_path=(
-                    f"{heading_segment.segment_path}/contains/"
-                    f"{segment.segment_path}"
+                    f"{heading_segment.segment_path}/contains/{segment.segment_path}"
                 ),
                 source_kind=segment.source_kind,
                 source_record_uid=segment.source_record_uid,
@@ -905,6 +952,8 @@ async def _generate_import_embeddings(
     embedding_provider: EmailImportEmbeddingProvider | None,
     batch_context: "EmailImportBatchContext | None" = None,
 ) -> list[list[float]]:
+    if not texts:
+        return []
     if embedding_provider is None:
         return [_zero_embedding() for _ in texts]
     if batch_context is not None and texts:
@@ -921,6 +970,21 @@ async def _generate_import_embeddings(
             dimension=EMBEDDING_DIMENSION,
         )
         if batched is not None:
+            if isinstance(batched, BatchEmbeddingPartial):
+                remainder: list[list[float]] = []
+                for start in range(
+                    0, len(batched.pending_texts), MAX_EMBEDDING_CHUNKS_PER_WINDOW
+                ):
+                    remainder.extend(
+                        await _generate_import_embeddings(
+                            batched.pending_texts[
+                                start : start + MAX_EMBEDDING_CHUNKS_PER_WINDOW
+                            ],
+                            embedding_provider=embedding_provider,
+                            batch_context=None,
+                        )
+                    )
+                return [*batched.completed_vectors, *remainder]
             return batched
     try:
         provider_embeddings = await generate_embeddings(