From f5cddee937e0f19f2d9bf055b2605a471a73d4a6 Mon Sep 17 00:00:00 2001 From: Kronecker <2026692687@qq.com> Date: Fri, 16 Oct 2026 23:08:13 +0800 Subject: [PATCH 1/2] feat: export reproducible retrieval history snapshots --- README.md | 1 + backend/app/api/routes/retrieval.py | 23 ++ backend/app/schemas/exports.py | 19 ++ backend/app/services/exports.py | 201 ++++++++++++++++ backend/tests/test_exports.py | 342 ++++++++++++++++++++++++++++ docs/ARCHITECTURE.md | 2 + docs/RETRIEVAL_EXPORT_API.md | 97 ++++++++ docs/TASKS.md | 8 + frontend/package.json | 3 +- frontend/src/services/exports.ts | 19 ++ frontend/src/services/session.ts | 18 +- frontend/tests/session.test.mjs | 70 ++++++ 12 files changed, 800 insertions(+), 3 deletions(-) create mode 100644 backend/app/schemas/exports.py create mode 100644 backend/app/services/exports.py create mode 100644 backend/tests/test_exports.py create mode 100644 docs/RETRIEVAL_EXPORT_API.md create mode 100644 frontend/src/services/exports.ts create mode 100644 frontend/tests/session.test.mjs diff --git a/README.md b/README.md index 8f8c5d3..08c85f5 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,7 @@ Vite 通常使用 http://localhost:5173。 - GET/POST `/api/documents`、GET/PATCH/DELETE `/api/documents/{id}`、GET `/api/documents/{id}/history` - POST `/api/retrieval`：已核验知识的关键词检索 - GET `/api/retrieval/runs`、GET `/api/retrieval/runs/{run_id}`：检索历史和原始结果 +- POST `/api/retrieval/exports`：管理员按编号导出 JSONL 历史与可选准备上下文，见 [导出契约](docs/RETRIEVAL_EXPORT_API.md) - `/api/indexes`：管理员构建/查询切分快照、发布/查询当前版本、查询 chunks 和发布历史，详见 [接口契约](docs/INDEX_API.md) ### 阶段 2 接口约定 diff --git a/backend/app/api/routes/retrieval.py b/backend/app/api/routes/retrieval.py index 85a59dd..3af793d 100644 --- a/backend/app/api/routes/retrieval.py +++ b/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)): """只检索已核验知识；保存原始检索结果供复核，不调用大模型。""" diff --git a/backend/app/schemas/exports.py b/backend/app/schemas/exports.py new file mode 100644 index 0000000..9e7e2b1 --- /dev/null +++ b/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 diff --git a/backend/app/services/exports.py b/backend/app/services/exports.py new file mode 100644 index 0000000..ca8bb85 --- /dev/null +++ b/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 diff --git a/backend/tests/test_exports.py b/backend/tests/test_exports.py new file mode 100644 index 0000000..2e0858b --- /dev/null +++ b/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"] diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a19aa2c..7d2f5bf 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -37,6 +37,8 @@ FastAPI + Pydantic 新增 `retrieval_runs` 保存查询、模式、语料指纹和完整结果快照，支持分页读取及按 UUID 复核。零命中如实记录；当前依据修改后失效不会改写历史，但后续检索必须重新检查审核与父记录状态。响应耗时是本次真实读取、分词、排序及片段生成耗时，不含写日志或模型调用。 +管理员 `POST /api/retrieval/exports` 按明确编号导出四种模式的原保存响应，并可包含关联准备上下文；以一个 SQL 语句快照读取，分别标记条目/chunk 粒度。JSONL 首行记录版本、选择范围、条数和后续记录字节摘要；输出前校验、限制容量，不写新历史、不依赖当前发布或模型，不导出认证数据。生成用量和费用为未采集，完整规则见 [导出契约](RETRIEVAL_EXPORT_API.md)。 + 关键词基线仍按知识条目检索，显示片段最长 240 字符；独立的持久化 chunks 和文本索引版本见下方。该接口不读取 chunks；chunk 与向量/RRF 的独立接口见后文。全语料计算明确限制条目数及总字符数，超限报错，适用于首批小规模资料与后续对照实验准备，不能当作生产检索性能结论。 diff --git a/docs/RETRIEVAL_EXPORT_API.md b/docs/RETRIEVAL_EXPORT_API.md new file mode 100644 index 0000000..49dfa3a --- /dev/null +++ b/docs/RETRIEVAL_EXPORT_API.md @@ -0,0 +1,97 @@ +# 检索与上下文历史导出 + +本接口为后续实验保留当时的检索证据与配置，不重新检索、不加载 Embedding 或调用生成模型，也不生成质量分数。四种模式仍保持各自证据粒度，不能把条目与 chunk 混为同一评测样本。 + +## 请求与权限 + +`POST /api/retrieval/exports`，管理员会话、允许的 Origin 和 `X-CSRF-Token` 均为必需。普通用户 403，匿名/失效会话 401；错误 Origin 或 CSRF 为 403。继承 [认证契约](AUTH_API.md)，不会返回用户、会话或凭据字段。 + +```json +{ + "run_ids": ["00000000-0000-4000-8000-000000000001"], + "include_contexts": false +} +``` + +编号仅为格式示例，实际应先从管理员 `GET /api/retrieval/runs` 选择已存在的记录。`run_ids` 必须为 1–100 个不重复 UUID，大小写规范化后仍重复也拒绝；`include_contexts` 为严格布尔值，默认 false。额外字段、空列表和非法编号为 422。 + +明确选择避免导出无边界的全库记录。任一检索编号不存在即 404，不生成部分文件。可选上下文是这些检索记录关联的**全部已保存准备记录**，不含其他检索的上下文。 + +## 下载与 JSONL 契约 + +- 成功为 200，`Content-Type: application/x-ndjson; charset=utf-8`，`Content-Disposition: attachment; filename="nalati-retrieval-.jsonl"`。 +- 响应固定 `Cache-Control: no-store` 和 `X-Content-Type-Options: nosniff`；`X-Records-SHA256` 等于文件首行的 `records_sha256`。 +- UTF-8 编码，每行一个 JSON 对象，每行（包括末行）以 LF 结束；内容中的换行由 JSON 转义。不是 CSV，也不是一个 JSON 数组。 +- 首行为 `record_type=manifest`。其余先为检索记录，再为上下文记录，各组按数据库 `created_at,id` 升序。请求 UUID 列表的顺序不影响记录区顺序。 + +首行字段： + +| 字段 | 含义 | +|---|---| +| `schema_version` | 固定 `nalati-retrieval-export-v1` | +| `export_id`、`exported_at` | 本次导出 UUID 与带时区 UTC 时间 | +| `selection` | 排序后的检索 UUID 及是否包含上下文 | +| `counts` | 实际 `retrieval_runs` 与 `rag_contexts` 条数，不含首行 | +| `records_sha256` | **首行之后全部原始 UTF-8 字节**的 SHA-256，含每条末尾 LF | +| `snapshot_semantics` | 固定 `stored_history_not_current_corpus` | +| `generation_measurements` | `status=not_called`，`usage/cost/currency=null`；没有生成调用或费用测量，不能解释为零成本 | + +检索行有 `record_type=retrieval_run`、`id`、`ranking_unit` 和 `payload`。`payload` 保留原已保存响应，不重新计算或修改分数、耗时、文本、来源及版本： + +| `payload.mode` | `ranking_unit` | 保留的配置与证据 | +|---|---|---| +| `keyword` | `knowledge_item` | 条目 BM25、中文分词/标题权重、片段与来源版本 | +| `keyword_chunks` | `chunk` | 文本发布/切分/语料版本、正文 BM25、chunk 原文出处 | +| `vector` | `chunk` | 文本/向量发布、模型完整 revision、空间/维度与精确余弦配置 | +| `hybrid` | `chunk` | 同一 chunk 粒度、两路完整候选/排名/原始分数/贡献、RRF 与模型空间配置 | + +上下文行有 `record_type=rag_context`、`id` 和原保存的 `payload`；保留准备参数、证据、排除原因、提示词版本、请求体及请求摘要。`model_called=false`，即使 `status=prepared` 也不是模型回答。原字段详见 [RAG_CONTEXT_API.md](RAG_CONTEXT_API.md)。不导出数据库 `created_by` 或认证表数据。 + +## 快照、容量与失败 + +- 检索与可选上下文通过**一个 PostgreSQL SQL 语句快照**读取，没有跨页读取造成的漂移；执行后新增的上下文留到下一次导出。 +- 已修改、删除或重新发布的知识不会改写历史。导出成功只说明当时记录可读取，不说明它仍是当前有效发布或来源仍受支持。 +- 每次最多 100 条检索、100 条关联上下文，完整 UTF-8 文件（含 manifest）最多 8 MiB。超量为 413，不静默截断；某个检索自身关联超过 100 条准备记录时，本版可先不含上下文导出，不能借此声称已导出全部准备记录。 +- 输出前完成结构、编号/关联、数据库元数据和准备请求摘要校验；未知/缺失字段、非有限数值或关联异常为 409。它是历史格式一致性检查，不是来源真实性或事实正确性证明。 +- 服务端先完整验证并组装文件，再返回下载响应；校验、容量或数据库失败不会先发出部分 JSONL。此接口不写入新的检索/上下文记录，不重建索引。 +- 同一历史选择的记录区字节和摘要可重复；manifest 的导出 UUID/时间每次变化，因此不要比较整个文件的摘要。若新增上下文后再次启用 `include_contexts`，记录区也会变化。SHA-256 仅供字节完整性核对，不是审阅签名或质量指标。 + +`elapsed_ms` 沿用各模式原始计时范围，不是生成模型或完整浏览器交互耗时。模型、语料与参数不一致的记录不能只凭分数直接做实验对比。 + +## 调用与核对 + +已有管理员登录并按认证契约准备 Cookie 文件和当前 CSRF 后，可在仓库根目录调用。Cookie 文件、真实请求、JSONL 和实际日志均留在 Git 元数据目录或仓库外；下面的编号仍需替换成真实已保存的检索编号。 + +```bash +mkdir -p .git/exports +curl --fail-with-body --request POST http://127.0.0.1:8000/api/retrieval/exports \ + --cookie "$COOKIE_JAR" \ + --header 'Origin: http://localhost:5173' \ + --header "X-CSRF-Token: $CSRF_TOKEN" \ + --header 'Content-Type: application/json' \ + --data '{"run_ids":["00000000-0000-4000-8000-000000000001"],"include_contexts":true}' \ + --output .git/exports/history.jsonl +``` + +仅在 HTTP 200 后按 JSONL 读取文件。核对摘要时不应重新序列化记录： + +```python +import hashlib +import json +from pathlib import Path + +header, records = Path(".git/exports/history.jsonl").read_bytes().split(b"\n", 1) +manifest = json.loads(header) +assert manifest["schema_version"] == "nalati-retrieval-export-v1" +assert hashlib.sha256(records).hexdigest() == manifest["records_sha256"] +rows = [json.loads(line) for line in records.rstrip(b"\n").split(b"\n")] +assert len(rows) == sum(manifest["counts"].values()) +``` + +前端调用层 `services/exports.ts` 的 `exportRetrievalHistory` 返回 Blob、受限文件名和记录摘要，沿用同一 Cookie/CSRF、401 撤销及错误处理，不放入 localStorage。此阶段没有新增下载页面，页面入口由后续前端集成；调用层测试是模拟网络的契约测试，不冒充浏览器下载验收。 + +## 验证与撤回 + +新增专项覆盖混合模式原始快照、可选准备记录、无命中、软删除后历史、摘要稳定性、权限/CSRF、未知字段/关联损坏拒绝、字节/条数上限、并发新增上下文以及真实 HTTP 后端重启后的同一记录区。坐标 fixture 仅用于导出协议，不构成真实领域模型实验。专项命令 `pytest -q backend/tests/test_exports.py`：**30 passed**；完整 `pytest -q backend/tests`：**254 passed, 1 skipped**，skip 为显式启用的真实 Embedding 用例，本阶段没有重跑模型。`ruff check backend`、`ruff format --check backend` 通过；前端在 `frontend/` 执行 `npm run test:session`：**3 passed**，`npm run build` 通过。 + +数据库不新增迁移，继续为 `0007_rag_contexts`。撤回导出路由、服务、调用层即可停止新下载；原检索/上下文历史与已保存文件不变，不为回滚删除数据库记录。 diff --git a/docs/TASKS.md b/docs/TASKS.md index 269089f..2646915 100644 --- a/docs/TASKS.md +++ b/docs/TASKS.md @@ -99,3 +99,11 @@ - 修正页面联调发现的采集时间缺失、审核/文档导航入口和过期功能说明；后端审核/授权不放宽，不猜测历史资料采集时间。具体路径、HTTP 与浏览器证据分工及合成截图见 [BROWSER_ACCEPTANCE.md](BROWSER_ACCEPTANCE.md)。 - 独立 HTTP 检查确认匿名401、普通用户八类管理资源403、缺失CSRF/错误Origin403、退出后401；目录写入使旧发布409，重新发布后恢复真实检索；准备上下文仍为 model_called=false。合成数据和本机运行环境、原始记录、凭据不入库。 - 该轮验收收尾了T02的真实浏览器API连接、T05知识录入/审核与文本/向量页面核心流程、Issue #5登录/Cookie/CSRF/角色的主要浏览器路径。Docker、HTTPS、多浏览器、浏览器改密及历史元数据修订页面仍待验收；生成模型和领域评测继续延期。本轮 `pytest -q backend/tests` **224 passed, 1 skipped**，`ruff check`、`ruff format --check` 和前端 `npm run build` 均通过；常规 skip 为须显式启用的真实模型专项，本轮另外通过浏览器与独立 HTTP 实际运行了固定 BGE small。 + +## 2026-10-16 T07/T10：可复现历史导出 + +- 浏览器联调与录入修正经 [PR #29](https://gitea.kronecker.cc/saint/border-red-post-nalati/pulls/29) 真正合入 dev，已核对 head/merge 都在最新 dev 祖先链；再从最新 dev 创建唯一阶段分支 `feat/backend-retrieval-export`。 +- 新增管理员/Origin/CSRF 保护的 `POST /api/retrieval/exports`，按明确 UUID 导出 JSONL，保留四模式原保存的查询、命中、原始分数、两路候选、模型/语料/切分参数、来源和耗时；可选择关联的准备上下文。条目与 chunk 粒度分开，生成模型使用量和费用为未采集，不伪造生成调用或实验分数。 +- 一个 SQL 语句读取历史快照，保留旧版本而不重新检索；导出前完整校验，最多100条检索/100条上下文/8MiB，不静默截断，不返回部分文件。不新增数据库迁移或环境配置。 +- 前端仅增加下载类型与调用层，复用 Cookie/CSRF/错误处理，不新增页面；真实浏览器下载与实验展示留给后续集成。完整字段、摘要核对、复现与回滚见 [RETRIEVAL_EXPORT_API.md](RETRIEVAL_EXPORT_API.md)。 +- 专项测试 **30 passed**，包括真实 PostgreSQL 并发快照和 HTTP 重启导出；完整回归 **254 passed, 1 skipped**（skip 为须显式启用的真实 Embedding 专项），`ruff check` 与 `ruff format --check` 通过。前端会话下载契约 **3 passed**，类型与生产构建通过。测试使用合成资料与坐标 fixture，未重跑模型推理或调用生成模型。 diff --git a/frontend/package.json b/frontend/package.json index 533fa60..cb3a272 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -6,7 +6,8 @@ "scripts": { "dev": "vite", "build": "vue-tsc -b && vite build", - "preview": "vite preview" + "preview": "vite preview", + "test:session": "node --test tests/session.test.mjs" }, "dependencies": { "lucide-vue-next": "^0.468.0", diff --git a/frontend/src/services/exports.ts b/frontend/src/services/exports.ts new file mode 100644 index 0000000..628c049 --- /dev/null +++ b/frontend/src/services/exports.ts @@ -0,0 +1,19 @@ +import { apiDownload } from './session' +import type { RetrievalRunResult } from '../api' +import type { RagContext } from './rag' + +export type ExportSelection = { run_ids: string[]; include_contexts?: boolean } +export type ExportManifest = { + record_type: 'manifest'; schema_version: 'nalati-retrieval-export-v1'; export_id: string; exported_at: string; + selection: Required; counts: { retrieval_runs: number; rag_contexts: number }; records_sha256: string; + snapshot_semantics: 'stored_history_not_current_corpus'; + generation_measurements: { status: 'not_called'; usage: null; cost: null; currency: null } +} +export type ExportRecord = ExportManifest + | { record_type: 'retrieval_run'; id: string; ranking_unit: 'knowledge_item' | 'chunk'; payload: RetrievalRunResult } + | { record_type: 'rag_context'; id: string; payload: RagContext } + +// Returns bytes/metadata; callers explicitly choose whether and where to save the file. +export const exportRetrievalHistory = (payload: ExportSelection) => apiDownload( + '/api/retrieval/exports', { method: 'POST', body: JSON.stringify(payload) }, true, +) diff --git a/frontend/src/services/session.ts b/frontend/src/services/session.ts index 10d5753..95cfbe7 100644 --- a/frontend/src/services/session.ts +++ b/frontend/src/services/session.ts @@ -6,10 +6,10 @@ export class ApiError extends Error { export function clearSessionToken() { csrfToken = null } -export async function apiRequest(path: string, init: RequestInit = {}, protect = false): Promise { +async function request(path: string, init: RequestInit, protect: boolean, accept: string): Promise { if (protect && !csrfToken) await apiRequest('/api/auth/me') const headers = new Headers(init.headers) - headers.set('Accept', 'application/json') + headers.set('Accept', accept) if (init.body) headers.set('Content-Type', 'application/json') if (protect && csrfToken) headers.set('X-CSRF-Token', csrfToken) const response = await fetch(path, { ...init, credentials: 'include', headers }) @@ -19,6 +19,11 @@ export async function apiRequest(path: string, init: RequestInit = {}, protec const detail = body && typeof body === 'object' && 'detail' in body ? body.detail : null throw new ApiError(response.status, typeof detail === 'string' ? detail : `请求失败（${response.status}），请检查登录状态或填写内容。`) } + return response +} + +export async function apiRequest(path: string, init: RequestInit = {}, protect = false): Promise { + const response = await request(path, init, protect, 'application/json') if (response.status === 204) return undefined as T const data = await response.json() if (path === '/api/auth/login' || path === '/api/auth/me') { @@ -26,3 +31,12 @@ export async function apiRequest(path: string, init: RequestInit = {}, protec } return data as T } + +export type ApiDownload = { blob: Blob; filename: string; recordsSha256: string | null } + +export async function apiDownload(path: string, init: RequestInit = {}, protect = false): Promise { + const response = await request(path, init, protect, 'application/x-ndjson') + const disposition = response.headers.get('Content-Disposition') ?? '' + const filename = /^attachment; filename="([A-Za-z0-9._-]+)"$/.exec(disposition)?.[1] ?? 'download.jsonl' + return { blob: await response.blob(), filename, recordsSha256: response.headers.get('X-Records-SHA256') } +} diff --git a/frontend/tests/session.test.mjs b/frontend/tests/session.test.mjs new file mode 100644 index 0000000..05f3614 --- /dev/null +++ b/frontend/tests/session.test.mjs @@ -0,0 +1,70 @@ +import assert from 'node:assert/strict' +import { readFile } from 'node:fs/promises' +import test from 'node:test' +import ts from 'typescript' + +async function session(t, fetch_) { + const previous = globalThis.fetch + globalThis.fetch = fetch_ + t.after(() => { globalThis.fetch = previous }) + const source = await readFile(new URL('../src/services/session.ts', import.meta.url), 'utf8') + const { outputText } = ts.transpileModule(source, { compilerOptions: { target: ts.ScriptTarget.ES2022, module: ts.ModuleKind.ESNext } }) + return import(`data:text/javascript;base64,${Buffer.from(outputText).toString('base64')}#${t.name}`) +} + +test('JSON login and binary export share Cookie/CSRF transport without parsing JSONL as JSON', async t => { + const content = '{"record_type":"manifest"}\n{"query":"合成测试"}\n' + const calls = [] + const api = await session(t, async (path, init) => { + calls.push([path, init]) + if (path === '/api/auth/login') return Response.json({ csrf_token: 'fixture-csrf' }) + return new Response(content, { headers: { 'Content-Disposition': 'attachment; filename="fixture.jsonl"', 'X-Records-SHA256': 'a'.repeat(64) } }) + }) + await api.apiRequest('/api/auth/login', { method: 'POST', body: '{}' }) + const file = await api.apiDownload('/api/retrieval/exports', { method: 'POST', body: '{"run_ids":[]}' }, true) + assert.equal(await file.blob.text(), content) + assert.equal(file.filename, 'fixture.jsonl') + assert.equal(file.recordsSha256, 'a'.repeat(64)) + assert.equal(calls[1][1].credentials, 'include') + assert.equal(calls[1][1].headers.get('X-CSRF-Token'), 'fixture-csrf') + assert.equal(calls[1][1].headers.get('Content-Type'), 'application/json') + assert.equal(calls[1][1].headers.get('Accept'), 'application/x-ndjson') + assert.equal(calls[0][1].headers.get('Accept'), 'application/json') +}) + +test('download refreshes missing session token and preserves server authorization errors', async t => { + const calls = [] + const api = await session(t, async (path, init) => { + calls.push([path, init]) + if (path === '/api/auth/me') return Response.json({ csrf_token: 'refreshed-csrf' }) + return Response.json({ detail: '需要管理员权限' }, { status: 403 }) + }) + await assert.rejects(api.apiDownload('/api/retrieval/exports', { method: 'POST', body: '{}' }, true), error => { + assert.ok(error instanceof api.ApiError) + assert.equal(error.status, 403) + assert.equal(error.message, '需要管理员权限') + return true + }) + assert.equal(calls.length, 2) + assert.equal(calls[0][0], '/api/auth/me') + assert.equal(calls[1][1].headers.get('X-CSRF-Token'), 'refreshed-csrf') +}) + +test('401 download clears stale token and unsafe filename is not propagated', async t => { + let current = 'old-csrf' + let failed = false + const api = await session(t, async (path, init) => { + if (path === '/api/auth/login' || path === '/api/auth/me') return Response.json({ csrf_token: current }) + assert.equal(init.headers.get('X-CSRF-Token'), current) + if (!failed) { + failed = true; current = 'new-csrf' + return Response.json({ detail: '请先登录' }, { status: 401 }) + } + return new Response('fixture', { headers: { 'Content-Disposition': 'attachment; filename="../../fixture.jsonl"' } }) + }) + await api.apiRequest('/api/auth/login', { method: 'POST', body: '{}' }) + await assert.rejects(api.apiDownload('/api/retrieval/exports', { method: 'POST' }, true), error => error.status === 401) + const file = await api.apiDownload('/api/retrieval/exports', { method: 'POST' }, true) + assert.equal(file.filename, 'download.jsonl') + assert.equal(file.recordsSha256, null) +}) -- 2.43.0 From b31cfb5de6e3ccde1d60a694b6c57575cf2741e2 Mon Sep 17 00:00:00 2001 From: jhy <184825448@qq.com> Date: Fri, 16 Oct 2026 23:12:53 +0800 Subject: [PATCH 2/2] fix: prevent caching retrieval history exports --- backend/app/api/routes/retrieval.py | 1 + 1 file changed, 1 insertion(+) diff --git a/backend/app/api/routes/retrieval.py b/backend/app/api/routes/retrieval.py index 3af793d..c998180 100644 --- a/backend/app/api/routes/retrieval.py +++ b/backend/app/api/routes/retrieval.py @@ -42,6 +42,7 @@ def download_history(payload: ExportHistory, db: Session = Depends(get_db)): media_type="application/x-ndjson; charset=utf-8", headers={ "Content-Disposition": f'attachment; filename="nalati-retrieval-{export_id}.jsonl"', + "Cache-Control": "no-store", "X-Records-SHA256": digest, "X-Content-Type-Options": "nosniff", }, -- 2.43.0