fix(search): preserve full-document trigram storage

ARCHITECTURE.md

@@ -135,6 +135,17 @@ boundaries in the pure access-policy evaluator for platform operations, but it
 does not bypass data-region or consent denies; see
 `docs/operations/auth-key-management.md`.
 
+## Search storage repair boundary (Proposed)
+
+Naruon owns its four normalized PostgreSQL search expressions. The forward
+`0020_search_trigram_storage` candidate replaces whole-document GiST indexes
+with GIN without changing stored content, scope, score, or ranking SQL. GIN
+does not supply distance-only kNN acceleration; representative query latency
+and migration lock/build cost remain release gates, not assumed equivalence.
+RankWeave continues to own fusion and query normalization. See
+[ADR-0020](docs/adr/0020-full-document-trigram-storage.md) and the
+[PostgreSQL reproduction](docs/doctoring/search_trigram_storage.md).
+
 ## Local deployment boundary
 
 `docker-compose.yml` provides the blessed local stack: Postgres with pgvector,

CHANGELOG.md

@@ -1,4 +1,7 @@
 ## [Unreleased]
+- Proposed: repair storage failures for long email and document content while
+  retaining complete text and search scores. Search latency and deployment
+  validation remain required before this change is released.
 - Starlette `TestClient`의 기존 `httpx2==2.5.0` pin을 core 개발·테스트 의존성으로 승격하고, deprecated `httpx` fallback 경고 억제를 제거했습니다.
 - 긴 이메일·첨부 본문을 의미 단위 청크로 임베딩한 뒤 기존 email/attachment 벡터 계약으로 평균화하고, 청크 요청·벡터 누적을 제한된 창으로 처리합니다. OpenAI `text-embedding-3-*`에는 저장 차원(`1536`)을 직접 요청하도록 보강했습니다. 합성 메일 fixture 5건(70청크)과 provider 요청 계약으로 1,536차원 벡터 경로를 검증했으며, 실행 시 선택한 임베딩 제공자에 본문·파싱된 첨부 텍스트를 전송할 수 있습니다. 회사 기밀 데이터는 fixture·commit·PR·log에 포함하지 않습니다.
 - EmailDetail 테스트가 지원하지 않는 스레드 병합/분리 버튼을 `textContent`뿐 아니라 `aria-label`과 `title` 접근 가능 이름으로도 검출하도록 바꿔, 아이콘 전용 버튼 회귀를 놓치지 않습니다.

backend/alembic/versions/0020_search_trigram_storage.py

@@ -0,0 +1,55 @@
+"""Replace whole-document GiST leaf arrays with full-content GIN indexes.
+
+Keep migration 0010 immutable and preserve its normalization and index names.
+GIN supports trigram predicates, not distance-only kNN acceleration; unchanged
+ranking queries require measured performance evidence before this proposal lands.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0020_search_trigram_storage"
+down_revision = "0019_email_read_state_repair"
+
+_SEARCH_INDEX_DEFINITIONS = (
+    (
+        "ix_email_records_search_document_trgm",
+        "email_records",
+        "search_normalized_text(coalesce(subject, '') || ' ' || body)",
+    ),
+    (
+        "ix_email_attachments_content_trgm",
+        "email_attachments",
+        "search_normalized_text(content)",
+    ),
+    (
+        "ix_content_segments_safe_text_trgm",
+        "content_segments",
+        "search_normalized_text(safe_text_content)",
+    ),
+    (
+        "ix_project_graph_objects_search_document_trgm",
+        "project_graph_objects",
+        "search_normalized_text(title || ' ' || summary)",
+    ),
+)
+
+
+def upgrade() -> None:
+    """Rebuild the four owner indexes atomically without rewriting documents."""
+    for index_name, table_name, document_expression in _SEARCH_INDEX_DEFINITIONS:
+        op.drop_index(index_name, table_name=table_name, if_exists=True)
+        op.create_index(
+            index_name,
+            table_name,
+            [sa.literal_column(document_expression).label("search_document")],
+            postgresql_using="gin",
+            postgresql_ops={"search_document": "gin_trgm_ops"},
+        )
+
+
+def downgrade() -> None:
+    """Keep corrected indexes so application rollback preserves large records."""
+    # Reinstating GiST can fail once valid high-entropy documents are stored.
+    # A different index strategy needs its own forward, data-preserving repair.
+    return None

backend/services/hybrid_retrieval/retrieval_channels.py

@@ -7,10 +7,10 @@
 
 Lexical channels rank by pg_trgm word-similarity distance (``<->>``)
 over the SQL expression ``search_normalized_text(<document text>)``,
-which migration 0010_language_agnostic_search indexes with GiST
-trigram indexes. The expressions built here MUST stay textually
-identical to the indexed expressions, or PostgreSQL will not use the
-indexes. Character trigrams are language-agnostic: no per-language
+which migration 0020_search_trigram_storage indexes with full-content GIN
+trigram indexes. The expressions built here preserve the indexed normalization,
+but GIN does not accelerate this distance-only ordering: query performance is
+a separate rollout gate. Character trigrams are language-agnostic: no per-language
 tokenizer or ``to_tsvector`` configuration is involved (G6).
 
 Dense channels rank by pgvector cosine distance over the stored
@@ -106,7 +106,7 @@ def _lexical_scored_statement(
         normalized_query_expression, document_expression
     )
     # ``document <->> query`` = 1 - word_similarity(query, document);
-    # kNN-ordering form served by the GiST trigram indexes.
+    # Preserve exact ranking; GIN does not provide GiST kNN acceleration.
     lexical_distance = document_expression.op("<->>")(
         normalized_query_expression
     )

backend/tests/test_email_read_state_migration_postgres.py

@@ -1,25 +1,34 @@
-"""PostgreSQL regression coverage for 0011_email_read_state.
+"""PostgreSQL regression coverage for read-state and search-storage repairs.
 
 String-matching the revision file's source (test_alembic_migrations.py)
 cannot detect a destructive downgrade or prove the upgrade is actually
 idempotent -- both require running the real migration against a real
 database in each of the shapes it must handle.
 """
 
-import subprocess
+import hashlib
 import secrets
+import subprocess
 import sys
 import uuid
+from datetime import datetime, timezone
 from pathlib import Path
 
 import asyncpg
 import pytest
 from asyncpg.exceptions import InvalidAuthorizationSpecificationError, InvalidPasswordError
-from sqlalchemy import text
+from sqlalchemy import func, inspect, select, text, update
 from sqlalchemy.engine import make_url
-from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 from core.config import settings
+from db.models import (
+    Attachment,
+    ContentNodeRecord,
+    ContentSegmentRecord,
+    Email,
+    ProjectGraphObjectRecord,
+)
 
 pytestmark = pytest.mark.postgres
 
@@ -225,3 +234,206 @@ async def test_upgrade_head_repairs_a_database_already_stamped_past_0011(
     _run_migrations(fresh_database_url)
 
     assert await _column_exists(fresh_database_url, "email_records", "is_read") is True
+
+
+def _search_storage_rows(record_suffix: str):
+    """Build related historical records using the production mapper dependencies."""
+    message_id = f"<search-storage-{record_suffix}@example.com>"
+    content_hash = hashlib.sha256(b"historical quartz").hexdigest()
+    email = Email(
+        user_id="search-storage-user",
+        organization_id="search-storage-org",
+        message_id=message_id,
+        sender="sender@example.com",
+        subject="historical",
+        body="quartz",
+        date=datetime(2026, 9, 5, tzinfo=timezone.utc),
+    )
+    attachment = Attachment(
+        email=email,
+        filename=f"search-storage-{record_suffix}.txt",
+        content="historical quartz",
+    )
+    node = ContentNodeRecord(
+        email=email,
+        content_node_uid=f"search-node-{record_suffix}",
+        source_kind="email_body",
+        source_record_uid=message_id,
+        node_kind="document",
+        node_path="/document[1]",
+        ordinal_index=0,
+        safe_text_content="historical quartz",
+        content_hash=content_hash,
+    )
+    segment = ContentSegmentRecord(
+        email=email,
+        content_node=node,
+        content_segment_uid=f"search-segment-{record_suffix}",
+        source_kind="email_body",
+        source_record_uid=message_id,
+        segment_kind="paragraph",
+        segment_path="/document[1]/paragraph[1]",
+        ordinal_index=0,
+        safe_text_content="historical quartz",
+        content_hash=content_hash,
+        word_count=2,
+    )
+    project_object = ProjectGraphObjectRecord(
+        email=email,
+        primary_content_segment=segment,
+        object_uid=f"search-project-{record_suffix}",
+        user_id=email.user_id,
+        organization_id=email.organization_id,
+        workspace_id="workspace-search-storage-org",
+        object_type="requirement",
+        title="historical",
+        summary="quartz",
+        confidence=0.9,
+        source_segment_uids=[segment.content_segment_uid],
+        extractor_name="deterministic_reference",
+        extractor_version="test",
+    )
+    return {
+        "email_records": email,
+        "email_attachments": attachment,
+        "content_segments": segment,
+        "project_graph_objects": project_object,
+    }
+
+
+def _search_document_values(column_names, document: str):
+    """Split a complete test document across a surface's real storage columns."""
+    # The first SHA-256 word fits the project's 240-character title limit.
+    parts = document.split(" ", 1) if len(column_names) == 2 else [document]
+    values = dict(zip(column_names, parts, strict=True))
+    if "safe_text_content" in values:
+        values["content_hash"] = hashlib.sha256(document.encode()).hexdigest()
+        values["word_count"] = len(document.split())
+    return values
+
+
+@pytest.mark.parametrize(
+    ("surface", "column_names"),
+    [
+        ("email_records", ("subject", "body")),
+        ("email_attachments", ("content",)),
+        ("content_segments", ("safe_text_content",)),
+        ("project_graph_objects", ("title", "summary")),
+    ],
+    ids=["email", "attachment", "segment", "project"],
+)
+@pytest.mark.asyncio
+async def test_search_trigram_storage_forward_repair_preserves_large_documents(
+    fresh_database_url,
+    surface,
+    column_names,
+):
+    """Catch whole-document index overflow and destructive index rollback."""
+    _run_migrations(fresh_database_url, revision="0019_email_read_state_repair")
+    engine = create_async_engine(fresh_database_url)
+    session_factory = async_sessionmaker(engine, expire_on_commit=False)
+    historical_rows = _search_storage_rows("historical")
+    model = type(historical_rows[surface])
+    primary_key = inspect(model).primary_key[0]
+    columns = [getattr(model, name) for name in column_names]
+    search_document = columns[0]
+    if len(columns) == 2:
+        search_document = columns[0] + " " + columns[1]
+
+    async def assert_document(row_id, expected: str, tail_query: str):
+        """Verify complete stored bytes and the literal perfect tail-word score."""
+        async with engine.connect() as connection:
+            row = (
+                await connection.execute(
+                    select(
+                        *columns,
+                        func.word_similarity(
+                            func.search_normalized_text(tail_query),
+                            func.search_normalized_text(search_document),
+                        ),
+                    ).where(primary_key == row_id)
+                )
+            ).one()
+        assert " ".join(row[:-1]) == expected
+        # A complete final word has exactly the query's trigrams, so score = 1.
+        assert row[-1] == 1.0
+
+    try:
+        async with session_factory.begin() as session:
+            session.add_all(historical_rows.values())
+            await session.flush()
+            historical_id = inspect(historical_rows[surface]).identity[0]
+
+        _run_migrations(fresh_database_url)
+        await assert_document(historical_id, "historical quartz", "quartz")
+        async with engine.connect() as connection:
+            index_methods = list(
+                await connection.scalars(
+                    text(
+                        "SELECT access_method.amname FROM pg_index AS index_entry "
+                        "JOIN pg_class AS index_object ON index_object.oid = index_entry.indexrelid "
+                        "JOIN pg_class AS table_object ON table_object.oid = index_entry.indrelid "
+                        "JOIN pg_namespace AS table_schema ON table_schema.oid = table_object.relnamespace "
+                        "JOIN pg_am AS access_method ON access_method.oid = index_object.relam "
+                        "WHERE table_schema.nspname = 'public' AND table_object.relname = :table_name "
+                        "AND index_object.relname LIKE '%_trgm' AND index_entry.indisvalid"
+                    ),
+                    {"table_name": surface},
+                )
+            )
+        assert index_methods == ["gin"], "storage repair must retain a valid trigram index"
+
+        documents = {}
+        for tail_query in ("quartz", "zircon", "topaz"):
+            # Distinct digests exercise diverse trigram keys, not repeated text;
+            # the non-hex tail word occurs only beyond the 32 KiB boundary.
+            prefix = " ".join(
+                hashlib.sha256(f"{surface}:{tail_query}:{index}".encode()).hexdigest()
+                for index in range(1024)
+            )
+            assert len(prefix.encode()) > 32 * 1024
+            documents[tail_query] = f"{prefix} {tail_query}"
+
+        inserted_rows = _search_storage_rows("inserted")
+        for name, value in _search_document_values(
+            column_names, documents["quartz"]
+        ).items():
+            setattr(inserted_rows[surface], name, value)
+        async with session_factory.begin() as session:
+            session.add_all(inserted_rows.values())
+            await session.flush()
+            inserted_id = inspect(inserted_rows[surface]).identity[0]
+        await assert_document(inserted_id, documents["quartz"], "quartz")
+
+        async with engine.begin() as connection:
+            await connection.execute(
+                update(model)
+                .where(primary_key.in_([historical_id, inserted_id]))
+                .values(_search_document_values(column_names, documents["zircon"]))
+            )
+        for row_id in (historical_id, inserted_id):
+            await assert_document(row_id, documents["zircon"], "zircon")
+
+        _run_migrations(fresh_database_url)
+        for row_id in (historical_id, inserted_id):
+            await assert_document(row_id, documents["zircon"], "zircon")
+
+        _run_downgrade(fresh_database_url, "0019_email_read_state_repair")
+        for row_id in (historical_id, inserted_id):
+            await assert_document(row_id, documents["zircon"], "zircon")
+        # Downgrade must retain the corrected indexes and their ability to accept
+        # new large values, not reinstall the known failing GiST representation.
+        async with engine.begin() as connection:
+            await connection.execute(
+                update(model)
+                .where(primary_key.in_([historical_id, inserted_id]))
+                .values(_search_document_values(column_names, documents["topaz"]))
+            )
+        for row_id in (historical_id, inserted_id):
+            await assert_document(row_id, documents["topaz"], "topaz")
+
+        _run_migrations(fresh_database_url)
+        for row_id in (historical_id, inserted_id):
+            await assert_document(row_id, documents["topaz"], "topaz")
+    finally:
+        await engine.dispose()