663 lines
24 KiB
Python
663 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from modelforge_api.domain.evaluation import (
|
|
AdvisorRecommendationCreate,
|
|
CandidatePoolEntry,
|
|
EmbeddingMigrationCreate,
|
|
EvaluationCaseCreate,
|
|
EvaluationCaseResultCreate,
|
|
EvaluationComparisonCreate,
|
|
EvaluationRevisionCreate,
|
|
EvaluationRunComplete,
|
|
EvaluationRunCreate,
|
|
EvaluationSuiteCreate,
|
|
MigrationUpdate,
|
|
ModelComparisonCandidateCreate,
|
|
ModelComparisonCreate,
|
|
RankedResult,
|
|
RerankingCaseResultCreate,
|
|
RerankingRunComplete,
|
|
RerankingRunCreate,
|
|
RetrievalCandidatePoolCreate,
|
|
RetrievalPipelineIdentityCreate,
|
|
)
|
|
from modelforge_api.persistence.models import (
|
|
Base,
|
|
Capability,
|
|
CapabilityContract,
|
|
CapabilityDeployment,
|
|
EmbeddingMigration,
|
|
EmbeddingSpace,
|
|
EvaluationCase,
|
|
Project,
|
|
RetrievalCandidatePool,
|
|
)
|
|
from modelforge_api.services.evaluation import EvaluationError, EvaluationService
|
|
|
|
|
|
def _session() -> Session:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
def _suite(service: EvaluationService, project_id: UUID, relevant: UUID):
|
|
return service.create_suite(
|
|
EvaluationSuiteCreate(
|
|
project_id=project_id,
|
|
key="examplerag-retrieval",
|
|
name="ExampleRAG retrieval",
|
|
description="Reviewed local retrieval cases.",
|
|
revision=EvaluationRevisionCreate(
|
|
revision="v1",
|
|
dataset_revision="corpus-v1",
|
|
retrieval_settings={"hybrid": True},
|
|
cases=[
|
|
EvaluationCaseCreate(
|
|
case_key="known-answer",
|
|
query="Where is the answer?",
|
|
relevant_chunk_ids=[relevant],
|
|
label_provenance={"source": "manual-review"},
|
|
critical=True,
|
|
review_status="approved",
|
|
)
|
|
],
|
|
),
|
|
)
|
|
)
|
|
|
|
|
|
def _run(
|
|
service: EvaluationService,
|
|
project_id: UUID,
|
|
revision_id: UUID,
|
|
case_id: UUID,
|
|
relevant: UUID,
|
|
*,
|
|
target: str,
|
|
rank: int,
|
|
):
|
|
run = service.create_run(
|
|
EvaluationRunCreate(
|
|
project_id=project_id,
|
|
suite_revision_id=revision_id,
|
|
target_kind=target,
|
|
target_index_ref=f"index-{target}",
|
|
embedding_space_ref=f"space-{target}",
|
|
corpus_revision="corpus-v1",
|
|
retrieval_config_digest="a" * 64,
|
|
environment_fingerprint={"engine": "test-v1"},
|
|
)
|
|
)
|
|
ranked = [RankedResult(chunk_id=uuid4(), score=1.0 - index / 100) for index in range(rank - 1)]
|
|
ranked.append(RankedResult(chunk_id=relevant, score=0.5))
|
|
return service.complete_run(
|
|
run.id,
|
|
EvaluationRunComplete(
|
|
results=[
|
|
EvaluationCaseResultCreate(case_id=case_id, ranked_results=ranked, latency_ms=10.0)
|
|
]
|
|
),
|
|
)
|
|
|
|
|
|
def test_metrics_have_hand_calculable_values_and_per_case_evidence() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
session.add(project)
|
|
session.commit()
|
|
relevant = uuid4()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, relevant)
|
|
case_id = session.query(EvaluationCase.id).scalar()
|
|
assert case_id is not None
|
|
run = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case_id,
|
|
relevant,
|
|
target="current",
|
|
rank=2,
|
|
)
|
|
assert run.aggregate_metrics == pytest.approx(
|
|
{
|
|
"recall_at_5": 1.0,
|
|
"recall_at_10": 1.0,
|
|
"mrr": 0.5,
|
|
"ndcg_at_10": 1 / 1.584962500721156,
|
|
}
|
|
)
|
|
evidence = service.case_results(run.id)[0]
|
|
assert evidence.first_relevant_rank == 2
|
|
assert evidence.critical is True
|
|
assert len(evidence.ranked_results) == 2
|
|
|
|
|
|
def test_comparison_blocks_critical_regression() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
session.add(project)
|
|
session.commit()
|
|
relevant = uuid4()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, relevant)
|
|
case_id = session.query(EvaluationCase.id).scalar()
|
|
assert case_id is not None
|
|
baseline = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case_id,
|
|
relevant,
|
|
target="current",
|
|
rank=1,
|
|
)
|
|
candidate = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case_id,
|
|
relevant,
|
|
target="shadow",
|
|
rank=5,
|
|
)
|
|
comparison = service.compare(
|
|
EvaluationComparisonCreate(
|
|
baseline_run_id=baseline.id,
|
|
candidate_run_id=candidate.id,
|
|
)
|
|
)
|
|
assert comparison.comparability == "comparable"
|
|
assert comparison.regressed_cases == 1
|
|
assert comparison.critical_regressions == 1
|
|
assert comparison.promotion_eligibility == "not_eligible"
|
|
assert comparison.eligibility_evidence["promotion_performed"] is False
|
|
|
|
|
|
def test_suite_revision_and_cases_are_immutable() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
session.add(project)
|
|
session.commit()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, uuid4())
|
|
from modelforge_api.persistence.models import EvaluationSuiteRevision
|
|
|
|
revision = session.get(EvaluationSuiteRevision, suite.latest_revision_id)
|
|
assert revision is not None
|
|
revision.top_k = 20
|
|
with pytest.raises(ValueError, match="immutable approved fields"):
|
|
session.commit()
|
|
|
|
|
|
def _pool_entries(
|
|
relevant: UUID, relevant_rank: int = 5, count: int = 40
|
|
) -> list[CandidatePoolEntry]:
|
|
identifiers = [uuid4() for _ in range(count)]
|
|
identifiers[relevant_rank - 1] = relevant
|
|
return [
|
|
CandidatePoolEntry(
|
|
id=str(identifier),
|
|
document_id=str(uuid4()),
|
|
score=1.0 - index / 100,
|
|
content_sha256=hashlib.sha256(f"chunk-{identifier}".encode()).hexdigest(),
|
|
)
|
|
for index, identifier in enumerate(identifiers)
|
|
]
|
|
|
|
|
|
def test_frozen_candidate_pool_is_fingerprinted_content_free_and_immutable() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
session.add(project)
|
|
session.commit()
|
|
relevant = uuid4()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, relevant)
|
|
case = session.query(EvaluationCase).one()
|
|
request = RetrievalCandidatePoolCreate(
|
|
project_id=project.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
evaluation_case_id=case.id,
|
|
source_embedding_space="space-current",
|
|
source_index_ref="rag_dense_nomic_v1",
|
|
corpus_revision="corpus-v1",
|
|
retrieval_config_digest="a" * 64,
|
|
ordered_candidates=_pool_entries(relevant),
|
|
)
|
|
first = service.create_candidate_pool(request)
|
|
repeated = service.create_candidate_pool(request)
|
|
assert first.id == repeated.id
|
|
assert first.candidate_count == 40
|
|
assert all("text" not in candidate for candidate in first.ordered_candidates)
|
|
stored = session.get(RetrievalCandidatePool, first.id)
|
|
assert stored is not None
|
|
stored.source_index_ref = "mutated"
|
|
with pytest.raises(ValueError, match="immutable approved fields"):
|
|
session.commit()
|
|
|
|
|
|
def test_pipeline_identity_and_reranking_run_enforce_same_frozen_pool() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
capability = Capability(key="rag.reranking", description="Reranking")
|
|
session.add_all([project, capability])
|
|
session.flush()
|
|
contract = CapabilityContract(
|
|
capability_id=capability.id,
|
|
version=1,
|
|
input_schema={},
|
|
output_schema={},
|
|
contract={},
|
|
upgrade_class="behavioral",
|
|
)
|
|
session.add(contract)
|
|
session.flush()
|
|
deployment = CapabilityDeployment(
|
|
capability_contract_id=contract.id,
|
|
deployment_candidate_id=uuid4(),
|
|
embedding_space_id=None,
|
|
artifact_set_id=uuid4(),
|
|
runtime_profile_id=uuid4(),
|
|
compute_node_id=uuid4(),
|
|
accelerator_id=uuid4(),
|
|
channel="experiment",
|
|
status="candidate",
|
|
production=False,
|
|
fallback_policy={"allowed": False},
|
|
config_fingerprint="b" * 64,
|
|
provenance={"test": True},
|
|
rollback_policy={},
|
|
)
|
|
session.add(deployment)
|
|
session.commit()
|
|
relevant = uuid4()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, relevant)
|
|
case = session.query(EvaluationCase).one()
|
|
pool = service.create_candidate_pool(
|
|
RetrievalCandidatePoolCreate(
|
|
project_id=project.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
evaluation_case_id=case.id,
|
|
source_embedding_space="space-current",
|
|
source_index_ref="rag_dense_nomic_v1",
|
|
corpus_revision="corpus-v1",
|
|
retrieval_config_digest="a" * 64,
|
|
ordered_candidates=_pool_entries(relevant, count=39),
|
|
)
|
|
)
|
|
assert pool.candidate_count == 39
|
|
common = {
|
|
"project_id": project.id,
|
|
"embedding_space_ref": "space-current",
|
|
"sparse_config_digest": "c" * 64,
|
|
"fusion_config_digest": "d" * 64,
|
|
"configuration": {"rrf": "v1"},
|
|
}
|
|
control = service.create_pipeline_identity(RetrievalPipelineIdentityCreate(**common))
|
|
candidate = service.create_pipeline_identity(
|
|
RetrievalPipelineIdentityCreate(
|
|
**common,
|
|
reranker_deployment_id=deployment.id,
|
|
reranker_config_digest="e" * 64,
|
|
)
|
|
)
|
|
assert control.identity_digest != candidate.identity_digest
|
|
assert candidate.migration_class == "behavioral"
|
|
run = service.create_reranking_run(
|
|
RerankingRunCreate(
|
|
project_id=project.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
pipeline_identity_id=candidate.id,
|
|
control_pipeline_identity_id=control.id,
|
|
candidate_pool_ids=[pool.id],
|
|
corpus_revision="corpus-v1",
|
|
environment_fingerprint={"runtime": "offline"},
|
|
)
|
|
)
|
|
ranked_ids = [UUID(str(item["id"])) for item in pool.ordered_candidates[:10]]
|
|
ranked_ids.remove(relevant)
|
|
ranked_ids.insert(0, relevant)
|
|
completed = service.complete_reranking_run(
|
|
run.id,
|
|
RerankingRunComplete(
|
|
results=[
|
|
RerankingCaseResultCreate(
|
|
case_id=case.id,
|
|
candidate_pool_id=pool.id,
|
|
ranked_results=[
|
|
RankedResult(chunk_id=identifier, score=1.0 - rank / 100)
|
|
for rank, identifier in enumerate(ranked_ids)
|
|
],
|
|
retrieval_latency_ms=90,
|
|
rerank_latency_ms=25,
|
|
total_latency_ms=115,
|
|
)
|
|
]
|
|
),
|
|
)
|
|
assert completed.aggregate_metrics["mrr"] == 1.0
|
|
assert completed.latency_metrics["rerank_p95_ms"] == 25
|
|
evidence = service.reranking_case_results(run.id)[0]
|
|
assert evidence.first_relevant_rank == 1
|
|
assert evidence.critical is True
|
|
|
|
control_metrics = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case.id,
|
|
relevant,
|
|
target="current",
|
|
rank=2,
|
|
)
|
|
matrix = service.create_model_comparison(
|
|
ModelComparisonCreate(
|
|
project_id=project.id,
|
|
capability_contract_id=contract.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
current_run_id=control_metrics.id,
|
|
title="Current retrieval plus Qwen reranker",
|
|
candidates=[
|
|
ModelComparisonCandidateCreate(
|
|
candidate_key="nomic-plus-qwen-reranker",
|
|
label="nomic + Qwen3-Reranker-0.6B",
|
|
status="evaluated",
|
|
candidate_kind="retrieval_pipeline",
|
|
pipeline_identity_id=candidate.id,
|
|
latency_ms={"baseline_p95": 90.0, "p95": 115.0},
|
|
resource_evidence={
|
|
"runtime_compatible": True,
|
|
"gpu_fit": True,
|
|
"measured": True,
|
|
"stale": False,
|
|
"resident_vram_bytes": 2_000,
|
|
},
|
|
security_state={
|
|
"supply_chain_status": "verified",
|
|
"license_status": "approved",
|
|
},
|
|
provenance={"evidence_level": "A"},
|
|
)
|
|
],
|
|
)
|
|
)
|
|
candidate_evidence = matrix.candidates[0]
|
|
assert candidate_evidence["reranking_run_id"] == str(run.id)
|
|
assert candidate_evidence["metric_deltas"]["mrr"] == pytest.approx(0.8)
|
|
recommendation = service.recommend(
|
|
matrix.id,
|
|
AdvisorRecommendationCreate(candidate_key="nomic-plus-qwen-reranker"),
|
|
)
|
|
assert recommendation.target_kind == "retrieval_pipeline"
|
|
assert recommendation.current_pipeline_identity_id == control.id
|
|
assert recommendation.candidate_pipeline_identity_id == candidate.id
|
|
assert recommendation.candidate_deployment_id == deployment.id
|
|
assert recommendation.verdict == "KEEP_CURRENT_EMBEDDING_ADD_RERANKER_CANDIDATE"
|
|
assert recommendation.migration_impact == {
|
|
"class": "behavioral",
|
|
"requires_reindex": False,
|
|
}
|
|
|
|
with pytest.raises(EvaluationError, match="pool provenance"):
|
|
service.create_reranking_run(
|
|
RerankingRunCreate(
|
|
project_id=project.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
pipeline_identity_id=candidate.id,
|
|
control_pipeline_identity_id=control.id,
|
|
candidate_pool_ids=[pool.id],
|
|
corpus_revision="different-corpus",
|
|
environment_fingerprint={},
|
|
)
|
|
)
|
|
|
|
|
|
def test_migration_requires_preflight_and_complete_validation() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
session.add(project)
|
|
session.flush()
|
|
migration = EmbeddingMigration(
|
|
project_id=project.id,
|
|
source_embedding_space="legacy-observed",
|
|
target_embedding_space_id=uuid4(),
|
|
source_index_ref="current",
|
|
target_index_ref="shadow",
|
|
corpus_revision=hashlib.sha256(b"corpus").hexdigest(),
|
|
total_chunks=10,
|
|
completed_chunks=0,
|
|
failed_chunks=0,
|
|
retried_chunks=0,
|
|
batch_size=2,
|
|
concurrency=1,
|
|
priority="background",
|
|
preflight_evidence={},
|
|
progress_evidence={},
|
|
validation_evidence={},
|
|
operational_metrics={},
|
|
)
|
|
session.add(migration)
|
|
session.commit()
|
|
service = EvaluationService(session)
|
|
# PLANNED -> PREFLIGHT is valid; BACKFILLING then requires passed evidence.
|
|
item = service.update_migration(
|
|
migration.id,
|
|
MigrationUpdate(
|
|
status="preflight", completed_chunks=0, failed_chunks=0, retried_chunks=0
|
|
),
|
|
)
|
|
assert item.status == "preflight"
|
|
with pytest.raises(EvaluationError, match="preflight"):
|
|
service.update_migration(
|
|
migration.id,
|
|
MigrationUpdate(
|
|
status="backfilling", completed_chunks=0, failed_chunks=0, retried_chunks=0
|
|
),
|
|
)
|
|
item = service.update_migration(
|
|
migration.id,
|
|
MigrationUpdate(
|
|
status="backfilling",
|
|
completed_chunks=0,
|
|
failed_chunks=0,
|
|
retried_chunks=0,
|
|
preflight_evidence={"passed": True},
|
|
),
|
|
)
|
|
assert item.priority == "background"
|
|
item = service.cancel_migration(migration.id)
|
|
assert item.cancel_requested is True
|
|
|
|
|
|
def test_suite_cases_are_addressable_and_migration_start_is_explicit() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
space = EmbeddingSpace(
|
|
capability_contract_id=uuid4(),
|
|
artifact_set_id=uuid4(),
|
|
runtime_profile_id=uuid4(),
|
|
identity_digest="4" * 64,
|
|
dimension=1024,
|
|
normalized=True,
|
|
identity_facts={"distance_metric": "cosine"},
|
|
)
|
|
session.add_all([project, space])
|
|
session.commit()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, uuid4())
|
|
case = session.query(EvaluationCase).one()
|
|
|
|
definitions = service.suite_cases(suite.id, suite.latest_revision_id)
|
|
assert [item.id for item in definitions] == [case.id]
|
|
assert definitions[0].label_provenance == {"source": "manual-review"}
|
|
|
|
migration = service.create_migration(
|
|
project.id,
|
|
EmbeddingMigrationCreate(
|
|
source_embedding_space="legacy-observed",
|
|
target_embedding_space_id=space.id,
|
|
source_index_ref="current",
|
|
target_index_ref="shadow-explicit-start",
|
|
corpus_revision="corpus-v1",
|
|
total_chunks=1,
|
|
),
|
|
)
|
|
started = service.start_migration(migration.id)
|
|
assert started.status == "preflight"
|
|
assert started.priority == "background"
|
|
|
|
|
|
def test_model_comparison_and_advisor_hard_block_critical_regression() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
capability = Capability(key="rag.embedding", description="Dense embedding")
|
|
session.add_all([project, capability])
|
|
session.flush()
|
|
contract = CapabilityContract(
|
|
capability_id=capability.id,
|
|
version=1,
|
|
input_schema={},
|
|
output_schema={},
|
|
contract={},
|
|
upgrade_class="behavioral",
|
|
)
|
|
session.add(contract)
|
|
session.commit()
|
|
relevant = uuid4()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, relevant)
|
|
case_id = session.query(EvaluationCase.id).scalar()
|
|
assert case_id is not None
|
|
baseline = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case_id,
|
|
relevant,
|
|
target="current",
|
|
rank=1,
|
|
)
|
|
candidate = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case_id,
|
|
relevant,
|
|
target="shadow",
|
|
rank=2,
|
|
)
|
|
|
|
matrix = service.create_model_comparison(
|
|
ModelComparisonCreate(
|
|
project_id=project.id,
|
|
capability_contract_id=contract.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
current_run_id=baseline.id,
|
|
title="ExampleRAG embedding candidates",
|
|
candidates=[
|
|
ModelComparisonCandidateCreate(
|
|
candidate_key="qwen-retrieval",
|
|
label="Qwen retrieval-aware",
|
|
status="evaluated",
|
|
evaluation_run_id=candidate.id,
|
|
embedding_space="space-qwen",
|
|
latency_ms={"baseline_p95": 100.0, "p95": 150.0},
|
|
resource_evidence={
|
|
"runtime_compatible": True,
|
|
"gpu_fit": True,
|
|
"measured": True,
|
|
"stale": False,
|
|
"resident_vram_bytes": 1_000,
|
|
},
|
|
migration_impact={"class": "requires_reindex", "chunks": 597},
|
|
security_state={
|
|
"supply_chain_status": "verified",
|
|
"license_status": "approved",
|
|
},
|
|
provenance={"evidence_level": "A"},
|
|
)
|
|
],
|
|
)
|
|
)
|
|
assert matrix.candidates[0]["quality_metrics"]["recall_at_10"] == 1.0
|
|
assert matrix.candidates[0]["critical_regressions"] == 1
|
|
|
|
recommendation = service.recommend(
|
|
matrix.id, AdvisorRecommendationCreate(candidate_key="qwen-retrieval")
|
|
)
|
|
assert recommendation.verdict == "KEEP_CURRENT"
|
|
assert recommendation.confidence == "HIGH"
|
|
assert recommendation.evidence_level == "A"
|
|
assert "critical_regression" in recommendation.blockers
|
|
assert recommendation.migration_impact["class"] == "requires_reindex"
|
|
assert recommendation.policy_snapshot["critical_regression_hard_block"] is True
|
|
|
|
|
|
def test_advisor_requires_more_evidence_for_capacity_blocked_candidate() -> None:
|
|
with _session() as session:
|
|
project = Project(key="examplerag", name="ExampleRAG", description="test")
|
|
capability = Capability(key="rag.embedding", description="Dense embedding")
|
|
session.add_all([project, capability])
|
|
session.flush()
|
|
contract = CapabilityContract(
|
|
capability_id=capability.id,
|
|
version=1,
|
|
input_schema={},
|
|
output_schema={},
|
|
contract={},
|
|
upgrade_class="behavioral",
|
|
)
|
|
session.add(contract)
|
|
session.commit()
|
|
service = EvaluationService(session)
|
|
suite = _suite(service, project.id, uuid4())
|
|
case = session.query(EvaluationCase).one()
|
|
baseline = _run(
|
|
service,
|
|
project.id,
|
|
suite.latest_revision_id,
|
|
case.id,
|
|
UUID(case.relevant_chunk_ids[0]),
|
|
target="current",
|
|
rank=1,
|
|
)
|
|
matrix = service.create_model_comparison(
|
|
ModelComparisonCreate(
|
|
project_id=project.id,
|
|
capability_contract_id=contract.id,
|
|
suite_revision_id=suite.latest_revision_id,
|
|
current_run_id=baseline.id,
|
|
title="Blocked candidates",
|
|
candidates=[
|
|
ModelComparisonCandidateCreate(
|
|
candidate_key="qwen-4b",
|
|
label="Qwen3-Embedding-4B",
|
|
status="blocked",
|
|
blockers=["insufficient_vram"],
|
|
provenance={"evidence_level": "B"},
|
|
resource_evidence={"runtime_compatible": True, "gpu_fit": False},
|
|
security_state={
|
|
"supply_chain_status": "planned",
|
|
"license_status": "compatible",
|
|
},
|
|
)
|
|
],
|
|
)
|
|
)
|
|
recommendation = service.recommend(
|
|
matrix.id, AdvisorRecommendationCreate(candidate_key="qwen-4b")
|
|
)
|
|
assert recommendation.verdict == "REQUIRES_MORE_EVIDENCE"
|
|
assert recommendation.confidence == "LOW"
|
|
assert "local_project_evaluation_missing" in recommendation.blockers
|