feat: persist isolated vector snapshots and exact retrieval

backend/tests/test_migrations.py

@@ -16,7 +16,7 @@ def test_startup_requires_migration(database):
 
 
 def test_startup_rejects_previous_schema(database):
-    migrate(database, "0004_auth_sessions")
+    migrate(database, "0005_chunk_indexes")
     with pytest.raises(RuntimeError, match="alembic upgrade head"):
         with TestClient(create_app(database)):
             pass
@@ -40,10 +40,7 @@ def test_upgrade_preserves_original_rows_and_repeated_upgrade(database):
         assert row["collected_at"] is None
         assert row["verification_status"] == "pending"
         assert connection.scalar(text("SELECT count(*) FROM audit_events")) == 2
-        assert (
-            connection.scalar(text("SELECT version_num FROM alembic_version"))
-            == "0005_chunk_indexes"
-        )
+        assert connection.scalar(text("SELECT version_num FROM alembic_version")) == "0006_vectors"
         assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == []
     migrate(database, "0001_baseline", downgrade=True)
     with database.connect() as connection:
@@ -98,3 +95,29 @@ def test_chunk_index_downgrade_preserves_users_sessions_knowledge_and_retrieval(
     assert client.get("/api/auth/me").status_code == 200
     assert client.get("/api/indexes").json()["total"] == 0
     assert client.get("/api/indexes/active").json()["generation"] == 0
+
+
+def test_vector_downgrade_retains_text_publications_and_prior_retrieval_runs(
+    client, migrated_database
+):
+    from test_retrieval import add_item
+    from test_vectors import ToyProvider, text_version, vector_build, vector_publish, vector_search
+
+    from app.services.embedding_runtime import get_embedding_provider
+
+    client.app.dependency_overrides[get_embedding_provider] = ToyProvider
+    add_item(client)
+    text_ = text_version(client)
+    version = vector_build(client, text_)
+    vector_publish(client, version)
+    history = vector_search(client).json()
+    migrate(migrated_database, "0005_chunk_indexes", downgrade=True)
+    with migrated_database.connect() as c:
+        assert c.scalar(text("SELECT active_version_id FROM index_state")) == text_["id"]
+        assert c.scalar(text("SELECT count(*) FROM chunks")) == 1
+        assert c.scalar(text("SELECT count(*) FROM users")) == 1
+        assert c.scalar(text("SELECT count(*) FROM index_publications")) == 1
+        assert c.scalar(text("SELECT id FROM retrieval_runs")) == history["run_id"]
+    migrate(migrated_database)
+    assert client.get("/api/vector-indexes/active").json()["generation"] == 0
+    assert client.get(f"/api/retrieval/runs/{history['run_id']}").json() == history

backend/tests/test_runtime.py

@@ -17,7 +17,7 @@ from app.services.auth import PASSWORD_HASHER
 
 
 @contextmanager
-def server(engine, *, authenticate=True):
+def server(engine, *, authenticate=True, environment_overrides=None):
     with socket.socket() as socket_:
         socket_.bind(("127.0.0.1", 0))
         port = socket_.getsockname()[1]
@@ -26,6 +26,7 @@ def server(engine, *, authenticate=True):
         "DATABASE_URL": engine.url.render_as_string(hide_password=False),
         "APP_ENV": "test",
         "AUTH_COOKIE_SECURE": "false",
+        **(environment_overrides or {}),
     }
     if authenticate:
         with Session(engine) as db:

backend/tests/test_vector_runtime.py

@@ -0,0 +1,91 @@
+"""Explicit opt-in: actual pinned model -> HTTP -> PostgreSQL -> HTTP process restart."""
+
+import os
+
+import pytest
+from sqlalchemy import text
+from test_runtime import server
+
+from app.services.embedding import SMALL_ZH
+
+
+@pytest.mark.skipif(
+    os.environ.get("RUN_LOCAL_EMBEDDING_TEST") != "1",
+    reason="requires optional CPU runtime and offline pinned model cache",
+)
+def test_real_local_model_vectors_retrieval_and_http_restart(migrated_database):
+    options = {
+        "EMBEDDING_ENABLED": "true",
+        "EMBEDDING_MODEL": "small",
+        "EMBEDDING_CACHE_DIR": os.environ["EMBEDDING_CACHE_DIR"],
+        "HF_HUB_OFFLINE": "1",
+        "HF_HUB_DISABLE_TELEMETRY": "1",
+        "HF_HUB_DISABLE_IMPLICIT_TOKEN": "1",
+        "TOKENIZERS_PARALLELISM": "false",
+    }
+    query = {"query": "合成测试中交通安排应查阅哪份资料？"}
+    with server(migrated_database, environment_overrides=options) as client:
+        ids = []
+        for topic, body in [
+            ("交通", "【合成测试】占位交通资料介绍测试公交线路与班次的查询入口。"),
+            ("住宿", "【合成测试】占位住宿资料介绍测试房型与入住须知。"),
+            ("审核", "【合成测试】占位审核资料说明测试出处核验流程。"),
+        ]:
+            item = client.post(
+                "/api/knowledge",
+                json={
+                    "title": f"{topic}合成测试",
+                    "content": body,
+                    "source_url": "https://example.com/",
+                    "collected_at": "2026-10-16T00:00:00Z",
+                },
+            ).json()
+            response = client.post(
+                f"/api/knowledge/{item['id']}/review",
+                json={
+                    "expected_version": item["version"],
+                    "verification_status": "verified",
+                    "reviewer": "test-reviewer",
+                    "verification_notes": "仅为模型/HTTP/数据库持久化测试，不是事实资料",
+                },
+            )
+            assert response.status_code == 200
+            ids.append(item["id"])
+        text_ = client.post("/api/indexes", json={"chunk_size": 200, "overlap": 20}).json()
+        assert (
+            client.post(
+                f"/api/indexes/{text_['id']}/publish", json={"expected_generation": 0}
+            ).status_code
+            == 200
+        )
+        built = client.post(
+            "/api/vector-indexes", json={"text_index_version_id": text_["id"]}, timeout=60
+        )
+        assert built.status_code == 201, built.text
+        version = built.json()
+        assert version["space_id"] == SMALL_ZH.space_id and version["dimension"] == 512
+        published = client.post(
+            f"/api/vector-indexes/{version['id']}/publish", json={"expected_generation": 0}
+        )
+        assert published.status_code == 200
+        active = published.json()
+        response = client.post("/api/retrieval/vector", json=query, timeout=60)
+        assert response.status_code == 200, response.text
+        first = response.json()
+        assert {h["knowledge_id"] for h in first["results"]} == set(ids)
+        assert all(-1 <= h["score"] <= 1 for h in first["results"])
+        assert first["config"]["space"]["spec"]["revision"] == SMALL_ZH.revision
+    with migrated_database.connect() as c:
+        assert (
+            c.scalar(
+                text("SELECT count(*) FROM chunk_vectors WHERE public.vector_dims(embedding)=512")
+            )
+            == 3
+        )
+        assert c.scalar(text("SELECT count(*) FROM vector_publications")) == 1
+    with server(migrated_database, environment_overrides=options) as client:
+        assert client.get("/api/vector-indexes/active").json() == active
+        assert client.get(f"/api/retrieval/runs/{first['run_id']}").json() == first
+        response = client.post("/api/retrieval/vector", json=query, timeout=60)
+        assert response.status_code == 200, response.text
+        assert response.json()["results"] == first["results"]

backend/tests/test_vectors.py

@@ -0,0 +1,552 @@
+"""Toy vectors test database mathematics and isolation, never model/domain quality."""
+
+from dataclasses import replace
+from uuid import UUID, uuid4
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import func, select, text, update
+from sqlalchemy.exc import DBAPIError, OperationalError
+from sqlalchemy.orm import Session
+from test_indexing import build, publish
+from test_retrieval import add_item
+
+from app.models.indexing import Chunk
+from app.models.retrieval import RetrievalRun
+from app.models.vectors import ChunkVector, EmbeddingSpace, VectorPublication, VectorVersion
+from app.services import vectors
+from app.services.embedding import EmbeddingError, EmbeddingSpec
+from app.services.embedding_runtime import EmbeddingRuntime, get_embedding_provider
+from app.services.indexing import locked_state, publish_index
+
+
+class ToyProvider:
+    spec = EmbeddingSpec("test-only-coordinate-fixture", "a" * 40, 3)
+    callback = None
+
+    def encode(self, texts, *, query=False):
+        if self.callback:
+            self.callback(query)
+        return [[1.0, 0.0, 0.0] if query or "alpha" in t else [0.0, 1.0, 0.0] for t in texts]
+
+
+@pytest.fixture
+def provider(client):
+    value = ToyProvider()
+    client.app.dependency_overrides[get_embedding_provider] = lambda: value
+    return value
+
+
+def text_version(client, **kwargs):
+    version = build(client, chunk_size=100, overlap=0, **kwargs)
+    generation = client.get("/api/indexes/active").json()["generation"]
+    assert publish(client, version, generation).status_code == 200
+    return version
+
+
+def vector_build(client, version):
+    response = client.post("/api/vector-indexes", json={"text_index_version_id": version["id"]})
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+def vector_publish(client, version, generation=0):
+    return client.post(
+        f"/api/vector-indexes/{version['id']}/publish", json={"expected_generation": generation}
+    )
+
+
+def vector_search(client, **kwargs):
+    return client.post("/api/retrieval/vector", json={"query": "alpha query", **kwargs})
+
+
+def test_exact_cosine_original_chunks_category_and_mixed_history(client, provider):
+    first = add_item(client, content="alpha 占位正文", category="交通")
+    second = add_item(client, content="beta 占位正文", category="其他")
+    third = add_item(client, content="gamma 占位正文", category="其他")
+    add_item(client, status="pending", content="alpha 待审核不应进入")
+    text_ = text_version(client)
+    version = vector_build(client, text_)
+    assert (
+        version["status"] == "ready" and version["dimension"] == 3 and version["chunk_count"] == 3
+    )
+    assert vector_search(client).status_code == 409
+    assert vector_publish(client, version).json()["is_current"]
+    result = vector_search(client).json()
+    assert result["mode"] == "vector" and result["total_hits"] == 3
+    assert [h["knowledge_id"] for h in result["results"]] == [
+        first["id"],
+        second["id"],
+        third["id"],
+    ]
+    assert [h["score"] for h in result["results"]] == [1.0, 0.0, 0.0]
+    assert result["config"]["space"]["id"] == provider.spec.space_id
+    assert result["text_index_version_id"] == text_["id"]
+    assert result["vector_version_id"] == version["id"]
+    assert result["corpus_fingerprint"] == text_["corpus_fingerprint"]
+    assert result["elapsed_ms"] > 0
+    assert result["results"][0]["snippet"] == first["content"]
+    assert result["results"][0]["snippet_start"] == 0
+    assert result["results"][0]["snippet_end"] == len(first["content"])
+    assert result["results"][0]["source_url"] == "https://example.com/"
+    filtered = vector_search(client, category="其他", top_k=1).json()
+    assert filtered["corpus_size"] == 2 and filtered["results"][0]["knowledge_id"] == second["id"]
+    empty = vector_search(client, category="不存在分类").json()
+    assert empty["status"] == "no_evidence" and empty["results"] == []
+    keyword = client.post("/api/retrieval", json={"query": "alpha"}).json()
+    history = client.get("/api/retrieval/runs?page_size=100").json()
+    assert {r["mode"] for r in history["items"]} == {"keyword", "vector"}
+    assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result
+    assert client.get(f"/api/retrieval/runs/{keyword['run_id']}").json() == keyword
+    client.cookies.clear()
+    assert vector_search(client).status_code == 200
+    assert client.get("/api/retrieval/runs").status_code == 401
+
+
+def test_multiple_chunks_keep_original_offsets_and_order(client, provider):
+    item = add_item(client, content="alpha 占位正文。" * 45)
+    version = vector_build(client, text_version(client))
+    vector_publish(client, version)
+    hits = vector_search(client, top_k=20).json()["results"]
+    assert len(hits) > 1
+    assert [h["chunk_ordinal"] for h in hits] == list(range(len(hits)))
+    assert len({h["chunk_id"] for h in hits}) == len(hits)
+    for h in hits:
+        assert h["snippet"] == item["content"][h["snippet_start"] : h["snippet_end"]]
+
+
+def test_same_dimension_different_revision_is_a_separate_space(client, provider, migrated_database):
+    add_item(client)
+    text_ = text_version(client)
+    first = vector_build(client, text_)
+    vector_publish(client, first)
+    old_space = provider.spec.space_id
+    provider.spec = replace(provider.spec, revision="b" * 40)
+    assert vector_search(client).status_code == 409
+    second = vector_build(client, text_)
+    assert second["space_id"] != old_space and second["dimension"] == first["dimension"]
+    assert vector_publish(client, second, 1).status_code == 200
+    assert vector_search(client).json()["config"]["space"]["id"] == provider.spec.space_id
+    assert client.get("/api/vector-indexes/spaces").json()["total"] == 2
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(EmbeddingSpace)) == 2
+        assert db.scalar(select(func.count()).select_from(ChunkVector)) == 2
+
+
+@pytest.mark.parametrize("bad", ["dimension", "zero", "nan", "count", "overflow", "unavailable"])
+def test_invalid_or_failed_encode_never_persists_partial_vectors(
+    client, provider, migrated_database, bad
+):
+    add_item(client, content="alpha 占位")
+    text_ = text_version(client)
+    original = vector_build(client, text_)
+    vector_publish(client, original)
+    before = client.get("/api/vector-indexes/active").json()
+
+    def fail(texts, **kwargs):
+        if bad == "overflow":
+            raise EmbeddingError("token limit")
+        if bad == "unavailable":
+            raise OSError("private-path-must-not-appear")
+        return {
+            "dimension": [[1, 0]],
+            "zero": [[0, 0, 0]],
+            "nan": [[float("nan"), 0, 0]],
+            "count": [],
+        }[bad]
+
+    provider.encode = fail
+    response = client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]})
+    assert response.status_code == (503 if bad == "unavailable" else 422)
+    assert "private-path" not in response.text
+    assert client.get("/api/vector-indexes/active").json() == before
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(VectorVersion)) == 1
+        assert db.scalar(select(func.count()).select_from(ChunkVector)) == 1
+
+
+def test_build_and_query_encode_release_locks_but_recheck_catalog(
+    client, provider, migrated_database
+):
+    item = add_item(client, content="alpha 占位")
+    text_ = text_version(client)
+    original = vector_build(client, text_)
+    vector_publish(client, original)
+
+    def mutate(query):
+        with Session(migrated_database) as other:
+            other.execute(text("SET LOCAL lock_timeout = '100ms'"))
+            state = locked_state(other)  # Must succeed during both document and query inference.
+            state.catalog_revision += 1
+            other.execute(
+                text("UPDATE knowledge_items SET content=:value WHERE id=:id"),
+                {"value": "修订占位", "id": item["id"]},
+            )
+            other.commit()
+
+    provider.callback = mutate
+    response = client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]})
+    assert response.status_code == 409
+    assert not client.get("/api/vector-indexes/active").json()["is_current"]
+    assert vector_search(client).status_code == 409
+    # Restore an active text/vector snapshot, then race during query encoding too.
+    provider.callback = None
+    text_ = text_version(client)
+    replacement = vector_build(client, text_)
+    vector_publish(client, replacement, 1)
+    provider.callback = mutate
+    assert vector_search(client).status_code == 409
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(VectorVersion)) == 2
+        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0
+
+
+def test_query_rejects_publication_changed_during_encode(client, provider, migrated_database):
+    add_item(client)
+    text_ = text_version(client)
+    first, second = vector_build(client, text_), vector_build(client, text_)
+    vector_publish(client, first)
+
+    def switch(query):
+        if query:
+            with Session(migrated_database) as other:
+                vectors.publish_vectors(other, UUID(second["id"]), 1, second["created_by"])
+
+    provider.callback = switch
+    assert vector_search(client).status_code == 409
+    assert client.get("/api/vector-indexes/active").json()["active_version"]["id"] == second["id"]
+
+
+def test_text_switch_during_build_discards_new_vectors(client, provider, migrated_database):
+    add_item(client)
+    first = text_version(client)
+    replacement = build(client)
+
+    def switch(query):
+        with Session(migrated_database) as other:
+            publish_index(other, UUID(replacement["id"]), 1, replacement["created_by"])
+
+    provider.callback = switch
+    assert (
+        client.post("/api/vector-indexes", json={"text_index_version_id": first["id"]}).status_code
+        == 409
+    )
+    assert client.get("/api/vector-indexes").json()["total"] == 0
+
+
+def test_out_of_band_evidence_change_fails_closed(client, provider, migrated_database):
+    item = add_item(client)
+    text_ = text_version(client)
+    version = vector_build(client, text_)
+    vector_publish(client, version)
+    with migrated_database.begin() as c:
+        c.execute(
+            text("UPDATE knowledge_items SET content='SQL 更新占位' WHERE id=:id"),
+            {"id": item["id"]},
+        )
+    assert not client.get("/api/vector-indexes/active").json()["is_current"]
+    assert vector_search(client).status_code == 409
+    assert vector_publish(client, version, 1).status_code == 409
+
+
+def test_optimistic_publication_audit_and_rollback(client, provider, migrated_database):
+    add_item(client)
+    text_ = text_version(client)
+    first, second = vector_build(client, text_), vector_build(client, text_)
+    assert vector_publish(client, first).json()["generation"] == 1
+    assert vector_publish(client, second).status_code == 409
+    assert vector_publish(client, first, 1).json()["generation"] == 1
+    with migrated_database.begin() as c:
+        c.execute(
+            text(
+                "ALTER TABLE vector_publications ADD CONSTRAINT test_reject_publication CHECK (false) NOT VALID"
+            )
+        )
+    assert vector_publish(client, second, 1).status_code == 409
+    assert client.get("/api/vector-indexes/active").json()["active_version"]["id"] == first["id"]
+    assert client.get(f"/api/vector-indexes/{first['id']}").json()["status"] == "published"
+    assert client.get(f"/api/vector-indexes/{second['id']}").json()["status"] == "ready"
+    assert client.get("/api/vector-indexes/publications").json()["total"] == 1
+    with migrated_database.begin() as c:
+        c.execute(text("ALTER TABLE vector_publications DROP CONSTRAINT test_reject_publication"))
+    assert vector_publish(client, second, 1).json()["generation"] == 2
+    assert client.get(f"/api/vector-indexes/{first['id']}").json()["status"] == "superseded"
+
+
+def test_vector_insert_failure_rolls_back_entire_build(client, provider, migrated_database):
+    add_item(client)
+    text_ = text_version(client)
+    with migrated_database.begin() as c:
+        c.execute(
+            text(
+                "ALTER TABLE chunk_vectors ADD CONSTRAINT test_reject_vector CHECK (false) NOT VALID"
+            )
+        )
+    assert (
+        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
+        == 409
+    )
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(VectorVersion)) == 0
+        assert db.scalar(select(func.count()).select_from(ChunkVector)) == 0
+        assert db.scalar(select(func.count()).select_from(EmbeddingSpace)) == 0
+
+
+@pytest.mark.parametrize(
+    "column,value",
+    [
+        ("dimension", 2),
+        ("space_id", "0" * 64),
+        ("text_index_version_id", str(uuid4())),
+        ("embedding", [1, 0]),
+        ("embedding", [2, 0, 0]),
+        ("embedding", [float("nan"), 0, 0]),
+    ],
+)
+def test_postgresql_rejects_cross_scope_dimensions_and_nonunit_values(
+    client, provider, migrated_database, column, value
+):
+    add_item(client)
+    version = vector_build(client, text_version(client))
+    with Session(migrated_database) as db:
+        with pytest.raises(DBAPIError):
+            db.execute(
+                update(ChunkVector)
+                .where(ChunkVector.version_id == version["id"])
+                .values(**{column: value})
+            )
+            db.commit()
+        db.rollback()
+        row = db.scalar(select(ChunkVector))
+        assert row.space_id == provider.spec.space_id and row.dimension == 3
+
+
+def test_same_width_cross_space_and_cross_text_chunks_are_rejected_by_database(
+    client, provider, migrated_database
+):
+    add_item(client)
+    original_text = text_version(client)
+    first = vector_build(client, original_text)
+    provider.spec = replace(provider.spec, revision="c" * 40)
+    second = vector_build(client, original_text)
+    newer_text = text_version(client)
+    with Session(migrated_database) as db:
+        other_chunk = db.scalar(select(Chunk.id).where(Chunk.index_version_id == newer_text["id"]))
+        for fields in [{"space_id": second["space_id"]}, {"chunk_id": other_chunk}]:
+            with pytest.raises(DBAPIError):
+                db.execute(
+                    update(ChunkVector)
+                    .where(ChunkVector.version_id == first["id"])
+                    .values(**fields)
+                )
+                db.commit()
+            db.rollback()
+
+
+def test_incomplete_vector_set_cannot_be_published_or_queried(client, provider, migrated_database):
+    add_item(client)
+    version = vector_build(client, text_version(client))
+    vector_publish(client, version)
+    with migrated_database.begin() as c:
+        c.execute(text("DELETE FROM chunk_vectors"))
+    assert vector_publish(client, version, 1).status_code == 409
+    assert vector_search(client).status_code == 409
+
+
+def test_only_current_text_and_bounded_complete_snapshots_can_build(
+    client, provider, migrated_database, monkeypatch
+):
+    assert vector_search(client).status_code == 409
+    add_item(client)
+    unpublished = build(client)
+    assert (
+        client.post(
+            "/api/vector-indexes", json={"text_index_version_id": unpublished["id"]}
+        ).status_code
+        == 409
+    )
+    text_ = text_version(client)
+    assert (
+        client.post(
+            "/api/vector-indexes", json={"text_index_version_id": unpublished["id"]}
+        ).status_code
+        == 409
+    )
+    monkeypatch.setattr(vectors, "MAX_VECTOR_CHUNKS", 0)
+    assert (
+        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
+        == 503
+    )
+    monkeypatch.undo()
+    with migrated_database.begin() as c:
+        c.execute(text("DELETE FROM chunks"))
+    assert (
+        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
+        == 409
+    )
+
+
+def test_empty_reviewed_corpus_can_be_published_without_fabricated_results(client, provider):
+    version = vector_build(client, text_version(client))
+    assert version["chunk_count"] == 0
+    assert vector_publish(client, version).json()["is_current"]
+    result = vector_search(client).json()
+    assert (
+        result["status"] == "no_evidence" and result["results"] == [] and result["corpus_size"] == 0
+    )
+
+
+def test_metadata_and_build_require_admin_and_csrf(client, provider):
+    add_item(client)
+    version = vector_build(client, text_version(client))
+    assert (
+        client.post(
+            "/api/vector-indexes",
+            json={"text_index_version_id": version["text_index_version_id"]},
+            headers={"X-CSRF-Token": ""},
+        ).status_code
+        == 403
+    )
+    client.cookies.clear()
+    for method, path in [
+        ("GET", "/api/vector-indexes"),
+        ("GET", "/api/vector-indexes/active"),
+        ("GET", "/api/vector-indexes/spaces"),
+        ("GET", "/api/vector-indexes/publications"),
+        ("GET", f"/api/vector-indexes/{version['id']}"),
+        ("POST", "/api/vector-indexes"),
+        ("POST", f"/api/vector-indexes/{version['id']}/publish"),
+    ]:
+        assert client.request(method, path, json={}).status_code == 401
+    assert (
+        client.post(
+            "/api/auth/register", json={"username": "vector_user", "password": "test-password-123"}
+        ).status_code
+        == 201
+    )
+    auth = client.post(
+        "/api/auth/login", json={"username": "vector_user", "password": "test-password-123"}
+    ).json()
+    client.headers["X-CSRF-Token"] = auth["csrf_token"]
+    assert client.get("/api/vector-indexes").status_code == 403
+    assert client.post("/api/vector-indexes", json={}).status_code == 403
+
+
+def test_vector_publication_serializes_with_catalog_and_other_publishers(
+    client, provider, migrated_database
+):
+    add_item(client)
+    text_ = text_version(client)
+    first, second = vector_build(client, text_), vector_build(client, text_)
+    with Session(migrated_database) as owner, Session(migrated_database) as other:
+        locked_state(owner)
+        other.execute(text("SET LOCAL lock_timeout = '100ms'"))
+        with pytest.raises(OperationalError) as blocked:
+            vectors.publish_vectors(other, UUID(first["id"]), 0, first["created_by"])
+        assert blocked.value.orig.sqlstate == "55P03"
+        other.rollback()
+        vectors.publish_vectors(owner, UUID(first["id"]), 0, first["created_by"])
+        with pytest.raises(HTTPException) as stale:
+            vectors.publish_vectors(other, UUID(second["id"]), 0, second["created_by"])
+        assert stale.value.status_code == 409
+        other.rollback()
+    with Session(migrated_database) as db:
+        assert db.scalar(select(func.count()).select_from(VectorPublication)) == 1
+
+
+def test_lazy_runtime_is_offline_and_does_not_wait_when_busy(monkeypatch):
+    from app.services import embedding_runtime
+
+    calls = []
+
+    def load(spec, **kwargs):
+        calls.append(kwargs)
+        return ToyProvider()
+
+    monkeypatch.setattr(embedding_runtime, "LocalEmbeddingProvider", load)
+    runtime = EmbeddingRuntime(ToyProvider.spec, None, 8, 2)
+    assert calls == []
+    assert runtime.encode(["alpha"]) == [[1, 0, 0]]
+    assert calls[0]["offline"] is True
+    runtime.encode(["alpha"], query=True)
+    assert len(calls) == 1
+    with runtime._lock:
+        with pytest.raises(HTTPException) as busy:
+            runtime.encode(["alpha"])
+        assert busy.value.status_code == 503
+
+
+def test_disabled_runtime_is_explicit_and_keyword_baseline_remains_available(client):
+    assert vector_search(client).status_code == 503
+    assert client.post("/api/retrieval", json={"query": "那拉提"}).status_code == 200
+
+
+def test_different_width_spaces_coexist_and_queries_never_compare_across_widths(client, provider):
+    add_item(client)
+    text_ = text_version(client)
+    first = vector_build(client, text_)
+    vector_publish(client, first)
+    provider.spec = replace(provider.spec, dimension=2)
+    provider.encode = lambda texts, **kwargs: [[1, 0] for _ in texts]
+    second = vector_build(client, text_)
+    assert vector_publish(client, second, 1).status_code == 200
+    result = vector_search(client)
+    assert result.status_code == 200, result.text
+    assert result.json()["total_hits"] == 1 and result.json()["results"][0]["score"] == 1
+    assert client.get("/api/vector-indexes/spaces").json()["total"] == 2
+
+
+def test_new_text_publication_requires_corresponding_vectors_and_keeps_history(client, provider):
+    add_item(client)
+    old_text = text_version(client)
+    old = vector_build(client, old_text)
+    vector_publish(client, old)
+    history = vector_search(client).json()
+    new_text = text_version(client)
+    active = client.get("/api/vector-indexes/active").json()
+    assert not active["is_current"] and active["active_version"]["id"] == old["id"]
+    assert vector_search(client).status_code == 409
+    assert vector_publish(client, old, 1).status_code == 409
+    replacement = vector_build(client, new_text)
+    assert vector_publish(client, replacement, 1).json()["is_current"]
+    assert client.get(f"/api/retrieval/runs/{history['run_id']}").json() == history
+    assert client.get("/api/vector-indexes/publications").json()["total"] == 2
+
+
+def test_snapshot_text_corruption_cannot_be_encoded_published_or_served(
+    client, provider, migrated_database
+):
+    add_item(client)
+    text_ = text_version(client)
+    version = vector_build(client, text_)
+    vector_publish(client, version)
+    with migrated_database.begin() as c:
+        c.execute(text("UPDATE chunks SET text='篡改占位'"))
+    assert not client.get("/api/vector-indexes/active").json()["is_current"]
+    assert (
+        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
+        == 409
+    )
+    assert vector_publish(client, version, 1).status_code == 409
+    assert vector_search(client).status_code == 409
+
+
+def test_request_shape_pagination_and_missing_vector_version(client, provider):
+    assert (
+        client.post(
+            "/api/vector-indexes", json={"text_index_version_id": str(uuid4()), "embedding": [1, 0]}
+        ).status_code
+        == 422
+    )
+    assert vector_search(client, top_k=21).status_code == 422
+    assert vector_search(client, mode="keyword").status_code == 422
+    assert vector_search(client, query="   ").status_code == 422
+    assert client.get("/api/vector-indexes?page=0").status_code == 422
+    assert client.get(f"/api/vector-indexes/{uuid4()}").status_code == 404
+    assert client.get("/api/vector-indexes/spaces").json()["total"] == 0
+
+
+def test_blank_optional_cache_path_is_not_the_working_directory():
+    from app.core.config import Settings
+
+    assert Settings(embedding_cache_dir="", _env_file=None).embedding_cache_dir is None