GEWU/tests/transports/test_mcp_facet_label_promotion.py

from datetime import UTC, datetime

from gewu.adapters.openrouter_jev import JevAssessment
from gewu.adapters.sqlite_facet_labels import SQLiteFacetLabelIndex
from gewu.domain.contracts import (
    FacetLabelProposal,
    IngestReviewCreate,
    JevFacetResolution,
    JevFacetScore,
    Locator,
    ProjectFacetLabel,
)
from gewu.transports.mcp_server import _promote_jev_labels


class _FixedEmbeddings:
    dimensions = 3
    model_id = "test-embedding-model"
    artifact_sha256 = "a" * 64

    def embed_queries(self, texts: tuple[str, ...]) -> tuple[tuple[float, ...], ...]:
        return tuple((1.0, 0.0, 0.0) for _ in texts)

    def embed_passages(self, texts: tuple[str, ...]) -> tuple[tuple[float, ...], ...]:
        return tuple((1.0, 0.0, 0.0) for _ in texts)


class _ProjectRegistry:
    def __init__(self) -> None:
        self.added: list[ProjectFacetLabel] = []

    def add_facet_labels(self, project_id: str, labels: tuple[ProjectFacetLabel, ...]):
        assert project_id == "dome-performance"
        self.added.extend(labels)


def _review(candidate: str = "immersive sound field") -> IngestReviewCreate:
    return IngestReviewCreate(
        review_id="review-facet-promotion-1",
        project_id="dome-performance",
        run_id="run-facet-promotion-1",
        source_revision_id="source-revision-facet-promotion-1",
        extraction_id="b" * 64,
        locator=Locator(kind="text_span", block_id="block-1", start=0, end=5),
        excerpt="quote",
        lens="technical",
        summary="A reusable spatial-audio technique",
        relevance_rationale="This is a direct, reusable project technique.",
        candidate_facet_labels=(
            FacetLabelProposal(
                facet="tags",
                value=candidate,
                definition="Audio techniques that render sound in immersive spatial fields.",
            ),
        ),
    )


def _assessment(
    value: str = "immersive sound field",
    *,
    fit: float = 0.97,
    new: float = 0.96,
    confidence: float = 0.93,
) -> JevAssessment:
    return JevAssessment(
        relation_probability=0.98,
        lens="technical",
        lens_confidence=0.99,
        model="typesafe/jev-test",
        facet_scores=(JevFacetScore(facet="tags", value=value, probability=fit),),
        facet_resolutions=(
            JevFacetResolution(
                facet="tags",
                value=value,
                fit_probability=fit,
                selected_new_label=True,
                new_probability=new,
                choice_confidence=confidence,
                rank=1,
            ),
        ),
    )


def test_high_confidence_top_rank_new_label_is_added_and_embedded(
    tmp_path, monkeypatch
) -> None:
    monkeypatch.setenv("GEWU_FACET_LABEL_FIT_THRESHOLD", "0.90")
    monkeypatch.setenv("GEWU_FACET_LABEL_NEW_THRESHOLD", "0.90")
    monkeypatch.setenv("GEWU_FACET_LABEL_CONFIDENCE_THRESHOLD", "0.80")
    registry = _ProjectRegistry()
    embeddings = _FixedEmbeddings()
    label_index = SQLiteFacetLabelIndex(tmp_path / "labels.sqlite3")

    resolutions = _promote_jev_labels(
        item=_review(),
        assessment=_assessment(),
        project_id="dome-performance",
        project_registry=registry,
        facet_label_index=label_index,
        embedding_provider=embeddings,
        existing_catalog=(),
        semantic_search_ready=True,
    )

    assert len(registry.added) == 1
    assert registry.added[0].canonical_label == "immersive sound field"
    assert resolutions[0].rank == 1
    assert resolutions[0].added_to_project
    cached = label_index.closest(
        project_id="dome-performance",
        facet="tags",
        query="spatial audio rendering",
        labels=tuple(registry.added),
        embeddings=embeddings,
        limit=1,
    )
    assert cached[0].canonical_label == "immersive sound field"


def test_matching_existing_label_is_never_promoted_as_new(
    tmp_path, monkeypatch
) -> None:
    monkeypatch.setenv("GEWU_FACET_LABEL_FIT_THRESHOLD", "0.90")
    monkeypatch.setenv("GEWU_FACET_LABEL_NEW_THRESHOLD", "0.90")
    monkeypatch.setenv("GEWU_FACET_LABEL_CONFIDENCE_THRESHOLD", "0.80")
    registry = _ProjectRegistry()
    existing = ProjectFacetLabel(
        label_id="tag-spatial-audio",
        facet="tags",
        canonical_label="spatial audio",
        definition="Audio techniques for placing or rendering sound spatially.",
        created_at=datetime(2026, 1, 1, tzinfo=UTC),
    )

    resolutions = _promote_jev_labels(
        item=_review("spatial audio"),
        assessment=_assessment("spatial audio"),
        project_id="dome-performance",
        project_registry=registry,
        facet_label_index=SQLiteFacetLabelIndex(tmp_path / "labels.sqlite3"),
        embedding_provider=_FixedEmbeddings(),
        existing_catalog=(existing,),
        semantic_search_ready=True,
    )

    assert registry.added == []
    assert not resolutions[0].added_to_project