306 lines
11 KiB
Python
306 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from gewu.adapters.git_repository import GitCanonicalRepository
|
|
from gewu.adapters.sqlite_search import SearchIndexError, SQLiteSearchIndex
|
|
from gewu.adapters.sqlite_vectors import SQLiteVectorIndex
|
|
from gewu.application.identity import Principal
|
|
from gewu.application.search import CanonicalSearchService, HybridSearchIndex, SearchError
|
|
from gewu.domain.contracts import (
|
|
Confidence,
|
|
EdgeEndpoint,
|
|
NodeType,
|
|
Scope,
|
|
SearchRequest,
|
|
TypedEdge,
|
|
)
|
|
from tests.domain.proposal_fixtures import NODE
|
|
|
|
|
|
class FixtureEmbeddings:
|
|
dimensions = 2
|
|
model_id = "fixture-e5@v1"
|
|
artifact_sha256 = "b" * 64
|
|
|
|
def embed_queries(self, texts: tuple[str, ...]) -> tuple[tuple[float, ...], ...]:
|
|
return tuple((1.0, 0.0) for _ in texts)
|
|
|
|
def embed_passages(self, texts: tuple[str, ...]) -> tuple[tuple[float, ...], ...]:
|
|
return tuple(
|
|
(1.0, 0.0) if text.startswith("Semantic match") else (0.0, 1.0) for text in texts
|
|
)
|
|
|
|
|
|
def _record_path(node_id: str, project_id: str) -> str:
|
|
project_key = hashlib.sha256(project_id.encode()).hexdigest()
|
|
node_key = hashlib.sha256(node_id.encode()).hexdigest()
|
|
return f"projects/{project_key}/nodes/{node_key}.yaml"
|
|
|
|
|
|
def _repository(tmp_path: Path) -> GitCanonicalRepository:
|
|
repository = GitCanonicalRepository(tmp_path / "knowledge.git")
|
|
nodes = [
|
|
NODE.model_copy(
|
|
update={
|
|
"node_id": f"dome:alpha-{number:02}",
|
|
"title": f"Alpha candidate {number:02}",
|
|
"summary": f"Alpha lexical candidate {number:02}.",
|
|
"node_type": NodeType.FINDING,
|
|
"confidence": Confidence.MEDIUM,
|
|
}
|
|
)
|
|
for number in range(12)
|
|
]
|
|
nodes.extend(
|
|
(
|
|
NODE.model_copy(
|
|
update={
|
|
"node_id": "dome:semantic-match",
|
|
"title": "Semantic match for the intent",
|
|
"summary": "A close meaning with no shared query word.",
|
|
"node_type": NodeType.TECHNIQUE,
|
|
"confidence": Confidence.HIGH,
|
|
}
|
|
),
|
|
NODE.model_copy(
|
|
update={
|
|
"node_id": "dome:graph-related",
|
|
"title": "Connected nonlexical technique",
|
|
"summary": "A related result found through one typed edge.",
|
|
"node_type": NodeType.TECHNIQUE,
|
|
"confidence": Confidence.MEDIUM,
|
|
}
|
|
),
|
|
NODE.model_copy(
|
|
update={
|
|
"node_id": "video:semantic-match",
|
|
"project_id": "interactive-video",
|
|
"title": "Semantic match in another project",
|
|
"summary": "Must remain outside the requested project scope.",
|
|
"node_type": NodeType.TECHNIQUE,
|
|
"confidence": Confidence.HIGH,
|
|
}
|
|
),
|
|
)
|
|
)
|
|
edges = (
|
|
TypedEdge(
|
|
edge_id="dome:search-expansion-edge",
|
|
edge_type="compared_with",
|
|
source=EdgeEndpoint(kind="node", id="dome:alpha-00"),
|
|
target=EdgeEndpoint(kind="node", id="dome:graph-related"),
|
|
interpretation_rationale="Both notes inform the same audience-attendance question.",
|
|
),
|
|
TypedEdge(
|
|
edge_id="dome:search-cross-project-edge",
|
|
edge_type="compared_with",
|
|
source=EdgeEndpoint(kind="node", id="dome:alpha-00"),
|
|
target=EdgeEndpoint(kind="node", id="video:semantic-match"),
|
|
interpretation_rationale="Cross-project edge must not widen search scope.",
|
|
),
|
|
)
|
|
files = {
|
|
_record_path(node.node_id, node.project_id): yaml.safe_dump(
|
|
node.model_dump(mode="json")
|
|
).encode()
|
|
for node in nodes
|
|
}
|
|
files.update(
|
|
{
|
|
f"edges/{hashlib.sha256(edge.edge_id.encode()).hexdigest()}.yaml": yaml.safe_dump(
|
|
edge.model_dump(mode="json")
|
|
).encode()
|
|
for edge in edges
|
|
}
|
|
)
|
|
repository.publish_files(
|
|
expected_head=repository.head(),
|
|
files=files,
|
|
message="Add hybrid search fixtures",
|
|
index_directory=tmp_path / "publishing",
|
|
)
|
|
return repository
|
|
|
|
|
|
def test_hybrid_search_fuses_ranks_and_preserves_scope_and_filters(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
from gewu.application import search as search_module
|
|
|
|
monkeypatch.setattr(search_module, "_VECTOR_CANDIDATE_LIMIT", 1)
|
|
repository = _repository(tmp_path)
|
|
search_path = tmp_path / "search.sqlite3"
|
|
lexical = SQLiteSearchIndex(search_path, repository)
|
|
vectors = SQLiteVectorIndex(search_path)
|
|
hybrid = HybridSearchIndex(
|
|
lexical=lexical,
|
|
vectors=vectors,
|
|
embeddings=FixtureEmbeddings(),
|
|
)
|
|
lexical.configure_vector_generation(
|
|
extension_loader=vectors.load_extension,
|
|
builder=hybrid.build_vector_generation,
|
|
is_current=hybrid.vector_generation_is_current,
|
|
)
|
|
service = CanonicalSearchService(
|
|
index=hybrid,
|
|
project_ids_provider=lambda: ("dome-performance", "interactive-video"),
|
|
canonical_head=repository.head,
|
|
)
|
|
principal = Principal(
|
|
"reader", "reader-key", frozenset({"viewer"}), frozenset({"dome-performance"})
|
|
)
|
|
request = SearchRequest(
|
|
query="alpha",
|
|
scope=Scope(projects=("dome-performance",), include_shared=False),
|
|
limit=20,
|
|
budget_tokens=50_000,
|
|
)
|
|
|
|
response = service.search(request, principal=principal)
|
|
first_generation = vectors.active_generation()
|
|
assert first_generation is not None
|
|
assert lexical.path == vectors.path
|
|
assert first_generation.revision == response.index_revision
|
|
assert first_generation.node_count == 15
|
|
result_ids = [item["node"]["node_id"] for item in response.data]
|
|
assert response.retrieval_mode == "hybrid"
|
|
assert result_ids.index("dome:semantic-match") < result_ids.index("dome:alpha-11")
|
|
graph_hit = next(
|
|
item for item in response.data if item["node"]["node_id"] == "dome:graph-related"
|
|
)
|
|
assert graph_hit["retrieval_path"] == "graph_expansion"
|
|
assert graph_hit["expanded_from"] == "dome:alpha-00"
|
|
assert graph_hit["expansion_edge"]["edge_type"] == "compared_with"
|
|
assert not any(node_id.startswith("video:") for node_id in result_ids)
|
|
assert all("retrieval_score" in item for item in response.data)
|
|
|
|
first_page = service.search(request.model_copy(update={"limit": 5}), principal=principal)
|
|
assert first_page.next_cursor is not None
|
|
second_page = service.search(
|
|
request.model_copy(update={"limit": 5, "cursor": first_page.next_cursor}),
|
|
principal=principal,
|
|
)
|
|
first_ids = {item["node"]["node_id"] for item in first_page.data}
|
|
second_ids = {item["node"]["node_id"] for item in second_page.data}
|
|
assert len(first_ids) == len(second_ids) == 5
|
|
assert first_ids.isdisjoint(second_ids)
|
|
|
|
class ChangedModelEmbeddings(FixtureEmbeddings):
|
|
model_id = "fixture-e5@v2"
|
|
|
|
changed_hybrid = HybridSearchIndex(
|
|
lexical=lexical,
|
|
vectors=vectors,
|
|
embeddings=ChangedModelEmbeddings(),
|
|
)
|
|
changed_profile_service = CanonicalSearchService(
|
|
index=changed_hybrid,
|
|
project_ids_provider=lambda: ("dome-performance", "interactive-video"),
|
|
canonical_head=repository.head,
|
|
)
|
|
with pytest.raises(SearchError, match="cursor is invalid"):
|
|
changed_profile_service.search(
|
|
request.model_copy(update={"limit": 5, "cursor": first_page.next_cursor}),
|
|
principal=principal,
|
|
)
|
|
|
|
filtered = service.search(
|
|
request.model_copy(
|
|
update={"node_types": (NodeType.TECHNIQUE,), "confidences": (Confidence.HIGH,)}
|
|
),
|
|
principal=principal,
|
|
)
|
|
assert [item["node"]["node_id"] for item in filtered.data] == ["dome:semantic-match"]
|
|
assert filtered.retrieval_mode == "hybrid"
|
|
|
|
lexical.configure_vector_generation(
|
|
extension_loader=vectors.load_extension,
|
|
builder=changed_hybrid.build_vector_generation,
|
|
is_current=changed_hybrid.vector_generation_is_current,
|
|
)
|
|
changed_response = changed_profile_service.search(
|
|
request.model_copy(update={"limit": 5}), principal=principal
|
|
)
|
|
changed_generation = vectors.active_generation()
|
|
assert changed_response.retrieval_mode == "hybrid"
|
|
assert changed_generation is not None and changed_generation.model_id == "fixture-e5@v2"
|
|
assert vectors.generation(first_generation.generation_id) is not None
|
|
|
|
class FailingEmbeddings(FixtureEmbeddings):
|
|
model_id = "fixture-e5@failed-build"
|
|
|
|
def embed_passages(self, texts: tuple[str, ...]) -> tuple[tuple[float, ...], ...]:
|
|
raise RuntimeError("fixture model failure")
|
|
|
|
failing_hybrid = HybridSearchIndex(
|
|
lexical=lexical,
|
|
vectors=vectors,
|
|
embeddings=FailingEmbeddings(),
|
|
)
|
|
lexical.configure_vector_generation(
|
|
extension_loader=vectors.load_extension,
|
|
builder=failing_hybrid.build_vector_generation,
|
|
is_current=failing_hybrid.vector_generation_is_current,
|
|
)
|
|
degraded = CanonicalSearchService(
|
|
index=failing_hybrid,
|
|
project_ids_provider=lambda: ("dome-performance", "interactive-video"),
|
|
canonical_head=repository.head,
|
|
).search(request.model_copy(update={"limit": 5}), principal=principal)
|
|
assert degraded.retrieval_mode == "keyword"
|
|
assert any(item.code == "HYBRID_RETRIEVAL_DEGRADED" for item in degraded.warnings)
|
|
assert degraded.next_cursor is None
|
|
|
|
|
|
def test_fts_and_vector_publication_roll_back_as_one_transaction(tmp_path: Path) -> None:
|
|
repository = _repository(tmp_path)
|
|
search_path = tmp_path / "search.sqlite3"
|
|
lexical = SQLiteSearchIndex(search_path, repository)
|
|
vectors = SQLiteVectorIndex(search_path)
|
|
hybrid = HybridSearchIndex(
|
|
lexical=lexical,
|
|
vectors=vectors,
|
|
embeddings=FixtureEmbeddings(),
|
|
)
|
|
|
|
def fail_after_vector_publish(connection, revision, nodes) -> None:
|
|
hybrid.build_vector_generation(connection, revision, nodes)
|
|
raise sqlite3.OperationalError("simulated termination before index commit")
|
|
|
|
lexical.configure_vector_generation(
|
|
extension_loader=vectors.load_extension,
|
|
builder=fail_after_vector_publish,
|
|
is_current=hybrid.vector_generation_is_current,
|
|
)
|
|
with pytest.raises(SearchIndexError):
|
|
lexical.search(
|
|
query="alpha",
|
|
project_ids=("dome-performance",),
|
|
include_shared=False,
|
|
limit=5,
|
|
)
|
|
assert lexical.diagnostic_status()[0] is None
|
|
assert vectors.active_generation() is None
|
|
|
|
lexical.configure_vector_generation(
|
|
extension_loader=vectors.load_extension,
|
|
builder=hybrid.build_vector_generation,
|
|
is_current=hybrid.vector_generation_is_current,
|
|
)
|
|
recovered = hybrid.search(
|
|
query="alpha",
|
|
project_ids=("dome-performance",),
|
|
include_shared=False,
|
|
limit=5,
|
|
)
|
|
active = vectors.active_generation()
|
|
assert recovered.retrieval_mode == "hybrid"
|
|
assert active is not None and active.revision == recovered.index_revision
|