feat: export reproducible retrieval history snapshots

backend/app/api/routes/retrieval.py

@@ -1,12 +1,14 @@
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import Response
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.db.session import get_db
 from app.models import RetrievalRun
 from app.schemas.catalog import Page
+from app.schemas.exports import ExportHistory
 from app.schemas.hybrid import (
     ChunkKeywordRequest,
     ChunkKeywordResponse,
@@ -17,6 +19,7 @@ from app.schemas.retrieval import RetrievalRequest, RetrievalResponse
 from app.schemas.vectors import VectorRetrievalResponse
 from app.services.auth import require_admin
 from app.services.embedding_runtime import get_embedding_provider
+from app.services.exports import export_history
 from app.services.hybrid import retrieve_chunks, retrieve_hybrid
 from app.services.retrieval import retrieve
 
@@ -25,6 +28,26 @@ router = APIRouter(prefix="/retrieval", tags=["retrieval"])
 RunResponse = RetrievalResponse | VectorRetrievalResponse | ChunkKeywordResponse | HybridResponse
 
 
+@router.post(
+    "/exports",
+    response_class=Response,
+    responses={200: {"content": {"application/x-ndjson": {}}, "description": "JSONL 历史快照"}},
+    dependencies=[Depends(require_admin)],
+)
+def download_history(payload: ExportHistory, db: Session = Depends(get_db)):
+    """管理员按明确编号导出历史；沿用 Origin/CSRF，失败不返回部分文件。"""
+    content, export_id, digest = export_history(db, payload)
+    return Response(
+        content,
+        media_type="application/x-ndjson; charset=utf-8",
+        headers={
+            "Content-Disposition": f'attachment; filename="nalati-retrieval-{export_id}.jsonl"',
+            "X-Records-SHA256": digest,
+            "X-Content-Type-Options": "nosniff",
+        },
+    )
+
+
 @router.post("", response_model=RetrievalResponse)
 def keyword_retrieval(payload: RetrievalRequest, db: Session = Depends(get_db)):
     """只检索已核验知识；保存原始检索结果供复核，不调用大模型。"""

backend/app/schemas/exports.py

@@ -0,0 +1,19 @@
+"""Explicit, bounded selection of immutable experiment history."""
+
+from uuid import UUID
+
+from pydantic import Field, field_validator
+
+from app.schemas.catalog import Input
+
+
+class ExportHistory(Input):
+    run_ids: list[UUID] = Field(min_length=1, max_length=100)
+    include_contexts: bool = Field(False, strict=True)
+
+    @field_validator("run_ids")
+    @classmethod
+    def unique_ids(cls, value):
+        if len(value) != len(set(value)):
+            raise ValueError("检索记录编号不得重复")
+        return value

backend/app/services/exports.py

@@ -0,0 +1,201 @@
+"""Read stored snapshots in one SQL statement; finish validation before any download."""
+
+import hashlib
+from typing import Annotated
+from uuid import uuid4
+
+from fastapi import HTTPException
+from pydantic import Field, TypeAdapter, ValidationError
+from sqlalchemy import String, literal, select, union_all
+from sqlalchemy.orm import Session
+
+from app.models.catalog import utc_now
+from app.models.rag import RagContext
+from app.models.retrieval import RetrievalRun
+from app.schemas.exports import ExportHistory
+from app.schemas.hybrid import ChunkKeywordResponse, HybridResponse
+from app.schemas.rag import ContextRead
+from app.schemas.retrieval import RetrievalResponse
+from app.schemas.vectors import VectorRetrievalResponse
+from app.services.completions import canonical_json, request_body
+
+MAX_CONTEXTS = 100
+MAX_BYTES = 8 * 1024 * 1024
+SCHEMA_VERSION = "nalati-retrieval-export-v1"
+RUN = TypeAdapter(
+    Annotated[
+        RetrievalResponse | ChunkKeywordResponse | VectorRetrievalResponse | HybridResponse,
+        Field(discriminator="mode"),
+    ]
+)
+
+
+def invalid_history():
+    return HTTPException(409, "历史记录格式或关联不一致，未生成导出文件")
+
+
+def same_fields(raw, parsed):
+    """Reject unknown/missing fields instead of silently dropping snapshot metadata."""
+    if isinstance(parsed, dict):
+        return (
+            isinstance(raw, dict)
+            and raw.keys() == parsed.keys()
+            and all(same_fields(raw[key], value) for key, value in parsed.items())
+        )
+    if isinstance(parsed, list):
+        return (
+            isinstance(raw, list)
+            and len(raw) == len(parsed)
+            and all(same_fields(left, right) for left, right in zip(raw, parsed))
+        )
+    return True
+
+
+def history_rows(db: Session, ids: list[str], include_contexts: bool):
+    runs = select(
+        literal(0).label("record_order"),
+        RetrievalRun.id.label("id"),
+        RetrievalRun.id.label("run_id"),
+        RetrievalRun.created_at.label("created_at"),
+        RetrievalRun.query.label("query"),
+        RetrievalRun.mode.label("mode"),
+        RetrievalRun.corpus_fingerprint.label("corpus_fingerprint"),
+        RetrievalRun.response.label("payload"),
+    ).where(RetrievalRun.id.in_(ids))
+    if include_contexts:
+        contexts = select(
+            literal(1).label("record_order"),
+            RagContext.id.label("id"),
+            RagContext.retrieval_run_id.label("run_id"),
+            RagContext.created_at.label("created_at"),
+            literal(None, String).label("query"),
+            literal(None, String).label("mode"),
+            literal(None, String).label("corpus_fingerprint"),
+            RagContext.response.label("payload"),
+        ).where(RagContext.retrieval_run_id.in_(ids))
+        statement = union_all(runs, contexts)
+    else:
+        statement = runs
+    # One PostgreSQL statement snapshot, including optional preparations. No offset
+    # pagination, ORM auth rows, current corpus lookups, or model dependencies.
+    return list(
+        db.execute(
+            statement.order_by("record_order", "created_at", "id").limit(
+                len(ids) + MAX_CONTEXTS + 1
+            )
+        ).mappings()
+    )
+
+
+def checked_run(row):
+    raw = row["payload"]
+    try:
+        value = RUN.validate_json(canonical_json(raw), strict=True)
+        if (
+            not same_fields(raw, value.model_dump(mode="json"))
+            or str(value.run_id) != row["id"]
+            or value.mode != row["mode"]
+            or value.query != row["query"]
+            or value.corpus_fingerprint != row["corpus_fingerprint"]
+            or value.created_at != row["created_at"]
+            or (value.status == "ok") != bool(value.results)
+        ):
+            raise ValueError("run mismatch")
+    except (ValidationError, ValueError, TypeError) as exc:
+        raise invalid_history() from exc
+    return value
+
+
+def checked_context(row, runs):
+    raw = row["payload"]
+    try:
+        value = ContextRead.model_validate_json(canonical_json(raw), strict=True)
+        run = runs[row["run_id"]]
+        if (
+            not same_fields(raw, value.model_dump(mode="json"))
+            or str(value.id) != row["id"]
+            or str(value.retrieval_run_id) != row["run_id"]
+            or value.created_at != row["created_at"]
+            or value.query != run.query
+            or value.retrieval_mode != run.mode
+            or value.corpus_fingerprint != run.corpus_fingerprint
+            or value.text_index_version_id != run.text_index_version_id
+            or value.text_generation != run.text_generation
+            or value.catalog_revision != getattr(run, "catalog_revision", value.catalog_revision)
+            or value.vector_version_id != getattr(run, "vector_version_id", None)
+            or value.vector_generation != getattr(run, "generation", None)
+        ):
+            raise ValueError("context mismatch")
+        if value.status == "prepared":
+            expected = request_body(
+                value.query,
+                [evidence.model_dump(mode="json") for evidence in value.evidence],
+                value.completion,
+            )
+            if (
+                not value.evidence
+                or canonical_json(value.request_body) != canonical_json(expected)
+                or value.request_sha256
+                != hashlib.sha256(canonical_json(expected).encode()).hexdigest()
+            ):
+                raise ValueError("request mismatch")
+        elif value.request_body is not None or value.request_sha256 is not None or value.evidence:
+            raise ValueError("unexpected request")
+    except (ValidationError, ValueError, TypeError, KeyError, AttributeError) as exc:
+        raise invalid_history() from exc
+
+
+def export_history(db: Session, payload: ExportHistory):
+    ids = sorted(str(value) for value in payload.run_ids)
+    rows = history_rows(db, ids, payload.include_contexts)
+    context_count = sum(row["record_order"] == 1 for row in rows)
+    if context_count > MAX_CONTEXTS:
+        raise HTTPException(413, "关联上下文超过100条，请缩小检索记录选择范围")
+    run_rows = [row for row in rows if row["record_order"] == 0]
+    if {row["id"] for row in run_rows} != set(ids):
+        raise HTTPException(404, "部分检索记录不存在，未生成导出文件")
+    runs = {row["id"]: checked_run(row) for row in run_rows}
+    lines = []
+    size = 0
+    for row in rows:
+        if row["record_order"]:
+            checked_context(row, runs)
+            record = {"record_type": "rag_context", "id": row["id"], "payload": row["payload"]}
+        else:
+            record = {
+                "record_type": "retrieval_run",
+                "id": row["id"],
+                "ranking_unit": "knowledge_item" if row["mode"] == "keyword" else "chunk",
+                "payload": row["payload"],
+            }
+        try:
+            line = (canonical_json(record) + "\n").encode("utf-8")
+        except (TypeError, ValueError, UnicodeError) as exc:
+            raise invalid_history() from exc
+        size += len(line)
+        if size > MAX_BYTES:
+            raise HTTPException(413, "导出超过8 MiB，请缩小检索记录选择范围")
+        lines.append(line)
+    records = b"".join(lines)
+    digest = hashlib.sha256(records).hexdigest()
+    export_id = str(uuid4())
+    manifest = {
+        "record_type": "manifest",
+        "schema_version": SCHEMA_VERSION,
+        "export_id": export_id,
+        "exported_at": utc_now().isoformat(),
+        "selection": {"run_ids": ids, "include_contexts": payload.include_contexts},
+        "counts": {"retrieval_runs": len(run_rows), "rag_contexts": context_count},
+        "records_sha256": digest,
+        "snapshot_semantics": "stored_history_not_current_corpus",
+        "generation_measurements": {
+            "status": "not_called",
+            "usage": None,
+            "cost": None,
+            "currency": None,
+        },
+    }
+    content = (canonical_json(manifest) + "\n").encode("utf-8") + records
+    if len(content) > MAX_BYTES:
+        raise HTTPException(413, "导出超过8 MiB，请缩小检索记录选择范围")
+    return content, export_id, digest

backend/tests/test_exports.py

@@ -0,0 +1,342 @@
+"""Export historical protocol fixtures, not domain scores or actual model usage."""
+
+import copy
+import hashlib
+import json
+from datetime import timedelta
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import event, func, select
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.attributes import flag_modified
+from test_rag import prepare
+from test_retrieval import add_item, search
+from test_runtime import server
+from test_vectors import ToyProvider, text_version, vector_build, vector_publish
+
+from app.models.auth import User
+from app.models.catalog import utc_now
+from app.models.rag import RagContext
+from app.models.retrieval import RetrievalRun
+from app.services import exports
+from app.services.embedding_runtime import get_embedding_provider
+
+PATH = "/api/retrieval/exports"
+
+
+def download(client, runs, **fields):
+    return client.post(PATH, json={"run_ids": [r["run_id"] for r in runs], **fields})
+
+
+def lines(response):
+    assert response.status_code == 200, response.text
+    assert response.content.endswith(b"\n")
+    return [json.loads(line) for line in response.content.rstrip(b"\n").split(b"\n")]
+
+
+def chunk_run(client):
+    add_item(client, content="alpha query 合成占位正文")
+    text_version(client)
+    response = client.post("/api/retrieval/chunks", json={"query": "alpha"})
+    assert response.status_code == 200, response.text
+    return response.json()
+
+
+def test_mixed_modes_exact_snapshots_manifest_digest_and_optional_contexts(
+    client, migrated_database
+):
+    add_item(client, content="alpha query\n合成测试正文")
+    text_ = text_version(client)
+    client.app.dependency_overrides[get_embedding_provider] = ToyProvider
+    vector_publish(client, vector_build(client, text_))
+    runs = []
+    for path in ["", "/chunks", "/vector", "/hybrid"]:
+        response = client.post("/api/retrieval" + path, json={"query": "alpha\nquery"})
+        assert response.status_code == 200, response.text
+        runs.append(response.json())
+    context = prepare(client, runs[-1]["run_id"]).json()
+    client.app.dependency_overrides.pop(get_embedding_provider)
+    response = download(client, list(reversed(runs)))
+    records = lines(response)
+    manifest = records[0]
+    assert manifest["schema_version"] == "nalati-retrieval-export-v1"
+    assert manifest["counts"] == {"retrieval_runs": 4, "rag_contexts": 0}
+    assert manifest["selection"] == {
+        "run_ids": sorted(r["run_id"] for r in runs),
+        "include_contexts": False,
+    }
+    assert manifest["snapshot_semantics"] == "stored_history_not_current_corpus"
+    assert manifest["generation_measurements"] == {
+        "status": "not_called",
+        "usage": None,
+        "cost": None,
+        "currency": None,
+    }
+    data = response.content.split(b"\n", 1)[1]
+    assert hashlib.sha256(data).hexdigest() == manifest["records_sha256"]
+    assert response.headers["X-Records-SHA256"] == manifest["records_sha256"]
+    assert response.headers["content-type"] == "application/x-ndjson; charset=utf-8"
+    assert response.headers["content-disposition"] == (
+        f'attachment; filename="nalati-retrieval-{manifest["export_id"]}.jsonl"'
+    )
+    assert response.headers["cache-control"] == "no-store"
+    assert response.headers["x-content-type-options"] == "nosniff"
+    assert [r["payload"] for r in records[1:]] == runs
+    assert [r["ranking_unit"] for r in records[1:]] == ["knowledge_item", "chunk", "chunk", "chunk"]
+    repeated = download(client, runs)
+    assert repeated.content.split(b"\n", 1)[1] == data
+    assert lines(repeated)[0]["export_id"] != manifest["export_id"]
+    included = download(client, runs, include_contexts=True)
+    all_records = lines(included)
+    assert all_records[0]["counts"] == {"retrieval_runs": 4, "rag_contexts": 1}
+    assert all_records[-1] == {
+        "record_type": "rag_context",
+        "id": context["id"],
+        "payload": context,
+    }
+    for forbidden in [
+        b"created_by",
+        b"password_hash",
+        b"test_admin",
+        b"csrf_token",
+        b"nalati_session",
+    ]:
+        assert forbidden not in included.content
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 4
+        assert db.scalar(select(func.count()).select_from(RagContext)) == 1
+
+
+def test_missing_selection_is_atomic_and_no_evidence_is_exportable(client):
+    run = search(client, query="没有语料的合成问题")
+    assert run["status"] == "no_evidence"
+    missing = download(client, [run, {"run_id": str(uuid4())}])
+    assert missing.status_code == 404 and "Content-Disposition" not in missing.headers
+    assert "manifest" not in missing.text
+    assert lines(download(client, [run]))[1]["payload"] == run
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {},
+        {"run_ids": []},
+        {"run_ids": ["not-a-uuid"]},
+        {"run_ids": [str(uuid4()) for _ in range(101)]},
+        {"run_ids": [str(uuid4())], "include_contexts": "true"},
+        {"run_ids": [str(uuid4())], "api_key": "test-secret-not-echoed"},
+    ],
+)
+def test_selection_validation_rejects_unsafe_or_unbounded_input(client, payload):
+    response = client.post(PATH, json=payload)
+    assert response.status_code == 422
+    assert "test-secret-not-echoed" not in response.text
+
+
+def test_duplicate_uuids_are_rejected_after_normalization(client):
+    run = search(client)
+    response = client.post(PATH, json={"run_ids": [run["run_id"], run["run_id"].upper()]})
+    assert response.status_code == 422
+
+
+def test_admin_origin_csrf_and_no_auth_data_in_errors(anonymous_client, migrated_database):
+    payload = {"run_ids": [str(uuid4())]}
+    assert anonymous_client.post(PATH, json=payload).status_code == 401
+    credentials = {"username": "export_user", "password": "test-password-123"}
+    assert anonymous_client.post("/api/auth/register", json=credentials).status_code == 201
+    login = anonymous_client.post("/api/auth/login", json=credentials).json()
+    anonymous_client.headers["X-CSRF-Token"] = login["csrf_token"]
+    assert anonymous_client.post(PATH, json=payload).status_code == 403
+    with Session(migrated_database) as db:
+        user = db.scalar(select(User).where(User.username == credentials["username"]))
+        user.role = "admin"
+        db.commit()
+    token = anonymous_client.headers.pop("X-CSRF-Token")
+    assert anonymous_client.post(PATH, json=payload).status_code == 403
+    anonymous_client.headers["X-CSRF-Token"] = token
+    assert (
+        anonymous_client.post(
+            PATH, json=payload, headers={"Origin": "https://example.com"}
+        ).status_code
+        == 403
+    )
+    assert anonymous_client.post(PATH, json=payload).status_code == 404
+
+
+def test_export_retains_history_after_catalog_deletion_and_context_is_no_generation(client):
+    add_item(client, content="alpha query 合成正文")
+    text_version(client)
+    run = client.post("/api/retrieval/chunks", json={"query": "不存在的词"}).json()
+    context = prepare(client, run["run_id"]).json()
+    assert context["status"] == "insufficient_evidence"
+    item = client.get("/api/knowledge").json()[0]
+    assert (
+        client.delete(f"/api/knowledge/{item['id']}?expected_version={item['version']}").status_code
+        == 204
+    )
+    result = lines(download(client, [run], include_contexts=True))
+    assert result[1]["payload"] == run and result[2]["payload"] == context
+    assert result[2]["payload"]["model_called"] is False
+
+
+@pytest.mark.parametrize(
+    "change", ["id", "mode", "query", "fingerprint", "created_at", "unknown", "score_type"]
+)
+def test_corrupt_run_fails_without_partial_download_or_private_error(
+    client, migrated_database, change
+):
+    add_item(client, content="alpha 合成正文")
+    run = search(client, query="alpha")
+    with Session(migrated_database) as db:
+        row = db.get(RetrievalRun, run["run_id"])
+        raw = copy.deepcopy(row.response)
+        if change == "id":
+            raw["run_id"] = str(uuid4())
+        elif change == "mode":
+            row.mode = "vector"
+        elif change == "query":
+            row.query = "private-test-data"
+        elif change == "fingerprint":
+            row.corpus_fingerprint = "f" * 64
+        elif change == "created_at":
+            row.created_at += timedelta(seconds=1)
+        elif change == "unknown":
+            raw["api_key"] = "private-test-data"
+        else:
+            raw["results"][0]["score"] = "NaN"
+        row.response = raw
+        db.commit()
+    response = download(client, [run])
+    assert response.status_code == 409
+    assert (
+        "Content-Disposition" not in response.headers and "private-test-data" not in response.text
+    )
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        "request",
+        "request_numeric",
+        "hash",
+        "link",
+        "catalog_revision",
+        "model_called",
+        "creator_extra",
+    ],
+)
+def test_corrupt_context_fails_atomically(client, migrated_database, change):
+    run = chunk_run(client)
+    context = prepare(client, run["run_id"]).json()
+    with Session(migrated_database) as db:
+        row = db.get(RagContext, context["id"])
+        raw = copy.deepcopy(row.response)
+        if change == "request":
+            raw["request_body"]["api_key"] = "private-test-data"
+        elif change == "request_numeric":
+            # Python equality alone treats True == 1; the actual wire body/hash must not.
+            raw["request_body"]["n"] = True
+        elif change == "hash":
+            raw["request_sha256"] = "f" * 64
+        elif change == "link":
+            raw["retrieval_run_id"] = str(uuid4())
+        elif change == "catalog_revision":
+            raw["catalog_revision"] += 1
+        elif change == "model_called":
+            raw["model_called"] = True
+        else:
+            raw["created_by"] = "private-test-data"
+        row.response = raw
+        # SQLAlchemy's Python comparison otherwise considers JSON true equal to 1.
+        flag_modified(row, "response")
+        db.commit()
+        if change == "request_numeric":
+            db.refresh(row)
+            assert row.response["request_body"]["n"] is True
+    response = download(client, [run], include_contexts=True)
+    assert response.status_code == 409 and "Content-Disposition" not in response.headers
+    assert "private-test-data" not in response.text
+    assert download(client, [run]).status_code == 200
+
+
+@pytest.mark.parametrize("boundary", ["record", "manifest"])
+def test_utf8_byte_cap_rejects_whole_file_including_manifest(client, monkeypatch, boundary):
+    run = search(client, query="多字节中文合成问题")
+    original = download(client, [run])
+    full_size = len(original.content)
+    record_size = len(original.content.split(b"\n", 1)[1])
+    monkeypatch.setattr(
+        exports, "MAX_BYTES", record_size - 1 if boundary == "record" else full_size - 1
+    )
+    response = download(client, [run])
+    assert response.status_code == 413 and "Content-Disposition" not in response.headers
+    assert client.get(f"/api/retrieval/runs/{run['run_id']}").json() == run
+    monkeypatch.setattr(exports, "MAX_BYTES", full_size)
+    assert len(download(client, [run]).content) == full_size
+
+
+def copied_context(context, run_id, actor):
+    raw = copy.deepcopy(context)
+    raw.update(id=str(uuid4()), created_at=utc_now().isoformat())
+    return RagContext(
+        id=raw["id"],
+        retrieval_run_id=run_id,
+        created_by=actor,
+        response=raw,
+        created_at=raw["created_at"],
+    )
+
+
+def test_context_capacity_rejects_instead_of_truncating(client, migrated_database):
+    run = chunk_run(client)
+    context = prepare(client, run["run_id"]).json()
+    with Session(migrated_database) as db:
+        actor = db.scalar(select(RagContext.created_by))
+        db.add_all(copied_context(context, run["run_id"], actor) for _ in range(100))
+        db.commit()
+    assert download(client, [run]).status_code == 200
+    response = download(client, [run], include_contexts=True)
+    assert response.status_code == 413 and "Content-Disposition" not in response.headers
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(RagContext)) == 101
+
+
+def test_one_statement_snapshot_excludes_concurrent_context_insert(client, migrated_database):
+    run = chunk_run(client)
+    context = prepare(client, run["run_id"]).json()
+    with Session(migrated_database) as db:
+        # Separate fixture actor avoids the authentication row lock held by the export request.
+        actor = User(username="snapshot_fixture", password_hash="not-a-login-hash", role="user")
+        db.add(actor)
+        db.commit()
+        actor_id = actor.id
+    inserted = False
+
+    def concurrent_insert(connection, cursor, statement, parameters, context_, many):
+        nonlocal inserted
+        if not inserted and "UNION ALL" in statement and "rag_contexts" in statement:
+            inserted = True
+            with Session(migrated_database) as db:
+                db.add(copied_context(context, run["run_id"], actor_id))
+                db.commit()
+
+    event.listen(migrated_database, "after_cursor_execute", concurrent_insert)
+    try:
+        result = lines(download(client, [run], include_contexts=True))
+    finally:
+        event.remove(migrated_database, "after_cursor_execute", concurrent_insert)
+    assert inserted and result[0]["counts"]["rag_contexts"] == 1
+    assert lines(download(client, [run], include_contexts=True))[0]["counts"]["rag_contexts"] == 2
+
+
+def test_http_export_records_and_digest_survive_process_restart(migrated_database):
+    with server(migrated_database) as client:
+        run = search(client, query="HTTP导出合成问题")
+        first = download(client, [run])
+        assert first.status_code == 200
+    with server(migrated_database) as client:
+        second = download(client, [run])
+        assert second.status_code == 200
+        assert second.content.split(b"\n", 1)[1] == first.content.split(b"\n", 1)[1]
+        assert second.headers["X-Records-SHA256"] == first.headers["X-Records-SHA256"]