from __future__ import annotations import hashlib import uuid from datetime import UTC, datetime, timedelta import pytest from pydantic import ValidationError from sqlalchemy import create_engine, update from sqlalchemy.orm import Session from modelforge_api.domain.runtime import ( AgentRuntimeProbeComplete, AgentRuntimeProbeFailure, AgentRuntimeProbeProgress, CompatibilityAssessmentCreate, ExecutionApprovalCreate, RuntimeEnvironmentCreate, RuntimeProbeCreate, RuntimeProfileCreate, ) from modelforge_api.persistence.models import ( Accelerator, ArtifactLocation, ArtifactSet, ArtifactSetMember, Base, ComputeNode, Model, ModelArtifact, ModelRevision, StorageRoot, UpstreamSnapshot, ) from modelforge_api.services.registry import RegistryConflict from modelforge_api.services.runtime import RuntimeService from modelforge_api.settings import Settings @pytest.fixture def session() -> Session: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as value: yield value def setup_runtime(session: Session): now = datetime.now(UTC) model = Model( key="qwen-runtime", display_name="Qwen3 Embedding", upstream_provider="Qwen", upstream_source="Qwen/Qwen3-Embedding-0.6B", upstream_metadata={ "tags": ["qwen3", "safetensors", "sentence-transformers"], "pipeline_tag": "feature-extraction", "library_name": "sentence-transformers", }, local_metadata={}, interpretation_metadata={}, modalities=[], parameter_metadata={}, license_metadata={"status": "captured_unreviewed"}, lifecycle="candidate", ) node = ComputeNode( key="gpu_node", hostname="GPU Node", enabled=True, liveness_state="online", lab_eligible=True, agent_capabilities=["runtime.probe.v1", "runtime.health.v1", "runtime.unload.v1"], hardware_fingerprint="h" * 64, ) session.add_all([model, node]) session.flush() accelerator = Accelerator( compute_node_id=node.id, device_index=0, device_uuid="GPU-test", name="NVIDIA RTX", compute_capability_major=8, compute_capability_minor=9, total_vram_bytes=17_171_480_576, driver_version="575.64.05", cuda_version="12.9", status="active", ) root = StorageRoot( compute_node_id=node.id, name="models", path="/host/models", agent_path="/data/artifacts/model-registry", status="ready", writable=True, ) revision = ModelRevision( model_id=model.id, upstream_revision="main", resolved_commit_sha="a" * 40, metadata_snapshot={}, immutable_at=now, ) session.add_all([accelerator, root, revision]) session.flush() snapshot = UpstreamSnapshot( model_id=model.id, repository_id=model.upstream_source, requested_revision="main", resolved_commit_sha=revision.resolved_commit_sha, access_state="public", metadata_snapshot={}, card_metadata={}, security_metadata={}, stale_after=now + timedelta(hours=1), ) session.add(snapshot) session.flush() names = [ ("config.json", "configuration", "json", "1" * 64, 100), ("tokenizer.json", "tokenizer", "json", "2" * 64, 100), ("modules.json", "configuration", "json", "3" * 64, 100), ("1_Pooling/config.json", "configuration", "json", "4" * 64, 100), ("model.safetensors", "weights", "safetensors", "5" * 64, 1000), ] artifacts = [] for name, role, file_format, digest, size in names: artifact = ModelArtifact( revision_id=revision.id, filename=name, artifact_type=role, serialization_format=file_format, sha256=digest, size_bytes=size, status="verified", security_status="static_checks_passed_unapproved", license_status="captured_unreviewed", quarantined=False, verification_details={"risk_flags": []}, verified_at=now, immutable_at=now, ) session.add(artifact) session.flush() artifacts.append(artifact) artifact_set = ArtifactSet( revision_id=revision.id, snapshot_id=snapshot.id, variant_key="safetensors-default", label="Safetensors", selection_reason="test", selected_paths=[name for name, *_ in names], total_size_bytes=1400, file_count=5, availability="local", status="verified", completeness="complete", security_status="static_checks_passed_unapproved", license_status="captured_unreviewed", immutable_at=now, ) session.add(artifact_set) session.flush() parent = "repositories/Qwen--Qwen3-Embedding/" + revision.resolved_commit_sha for ordinal, artifact in enumerate(artifacts): session.add_all( [ ArtifactSetMember( artifact_set_id=artifact_set.id, artifact_id=artifact.id, ordinal=ordinal, required=True, ), ArtifactLocation( artifact_id=artifact.id, storage_root_id=root.id, relative_path=f"{parent}/{artifact.filename}", status="verified", size_bytes=artifact.size_bytes, observed_sha256=artifact.sha256, last_checked_at=now, ), ] ) session.commit() service = RuntimeService( session, Settings(database_url="sqlite+pysqlite:///:memory:", runtime_artifact_root="/models"), ) return service, artifact_set, node, artifacts def create_environment(service: RuntimeService, adapter: str = "sentence_transformers"): supported = ( ["qwen3"] if adapter in {"sentence_transformers", "qwen3_reranker", "transformers", "vllm"} else [] ) return service.create_environment( RuntimeEnvironmentCreate( name=f"{adapter} pinned", adapter=adapter, runtime_version="4.1.0", image_repository="modelforge-runtime-worker", image_digest="sha256:" + "a" * 64, python_version="3.12.11", cuda_runtime_version="12.8", package_versions={ "torch": "2.7.1", "transformers": "4.51.3", "sentence-transformers": "4.1.0", adapter: "4.1.0", "llama-cpp-python": "0.3.16", "diffusers": "0.35.1", }, supported_model_types=supported, supported_formats=["safetensors"], supported_modalities=["reranking" if adapter == "qwen3_reranker" else "embedding"], ) ) def create_profile( service: RuntimeService, artifact_set_id: uuid.UUID, adapter: str = "sentence_transformers" ): environment = create_environment(service, adapter) return service.create_profile( RuntimeProfileCreate( name=f"Qwen {adapter}", runtime_environment_id=environment.id, artifact_set_id=artifact_set_id, modality="reranking" if adapter == "qwen3_reranker" else "embedding", ) ) def prepare_probe(service: RuntimeService, artifact_set, node): profile = create_profile(service, artifact_set.id) assessment = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) approval = service.approve( artifact_set.id, ExecutionApprovalCreate( reason="Reviewed exact verified set for isolated lab execution", approved_by="test-operator", ), ) probe = service.create_probe( RuntimeProbeCreate( compatibility_assessment_id=assessment.id, execution_approval_id=approval.id, ) ) return profile, assessment, approval, probe def test_static_sentence_transformers_compatibility_is_reproducible(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) profile = create_profile(service, artifact_set.id) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert result.status == "compatible" assert result.artifact_facts["required_complete"] is True assert result.evidence["locally_proven_support"] is False assert result.evidence["resource_estimate"]["kind"] == "estimated" revision = service.repo.revision(artifact_set.revision_id) assert revision is not None filtered = service.assessments( model_id=revision.model_id, compute_node_id=node.id, runtime_profile_id=profile.id, status="compatible", ) assert [item.id for item in filtered] == [result.id] assert service.assessments(status="blocked") == [] def test_artifact_manifest_allows_distinct_paths_with_the_same_content_digest( session: Session, ) -> None: service, artifact_set, node, artifacts = setup_runtime(session) tokenizer = artifacts[1] existing_location = service.repo.artifact_locations([tokenizer.id])[0] duplicate_path = "tokenizer_config.json" selected_paths = [*artifact_set.selected_paths, duplicate_path] session.execute( update(ArtifactSet) .where(ArtifactSet.id == artifact_set.id) .values(selected_paths=selected_paths, file_count=len(selected_paths)) ) session.add( ArtifactLocation( artifact_id=tokenizer.id, storage_root_id=existing_location.storage_root_id, relative_path=existing_location.relative_path.rsplit("/", 1)[0] + f"/{duplicate_path}", status="verified", size_bytes=tokenizer.size_bytes, observed_sha256=tokenizer.sha256, last_checked_at=datetime.now(UTC), ) ) session.commit() profile = create_profile(service, artifact_set.id) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert result.status == "compatible" assert result.artifact_facts["required_complete"] is True assert result.artifact_facts["file_count"] == 6 assert result.artifact_facts["content_blob_count"] == 5 assert duplicate_path in result.artifact_facts["verified_location_paths"] approval = service.approve( artifact_set.id, ExecutionApprovalCreate(reason="Reviewed duplicate-content manifest", approved_by="test"), ) probe = service.create_probe( RuntimeProbeCreate( compatibility_assessment_id=result.id, execution_approval_id=approval.id, ) ) lease = service.claim_next(node) assert lease is not None assert lease.probe_id == probe.id manifest_paths = [item["path"] for item in lease.expected_manifest["files"]] assert manifest_paths == selected_paths @pytest.mark.parametrize( ("adapter", "blocker"), [("llama_cpp", "NO_GGUF_ARTIFACT_VARIANT"), ("diffusers", "NOT_A_DIFFUSION_PIPELINE")], ) def test_wrong_runtime_is_incompatible_by_artifact_evidence( session: Session, adapter: str, blocker: str ) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) profile = create_profile(service, artifact_set.id, adapter) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert result.status == "blocked" assert blocker in result.blockers @pytest.mark.parametrize("mutation", ["incomplete", "corrupt", "remote-code", "pickle", "no-gpu"]) def test_static_assessment_fails_closed(session: Session, mutation: str) -> None: service, artifact_set, node, artifacts = setup_runtime(session) if mutation == "incomplete": artifact_set.completeness = "partial" elif mutation == "corrupt": artifacts[0].status = "corrupt" elif mutation == "remote-code": artifacts[0].verification_details = {"risk_flags": ["remote_code"]} elif mutation == "pickle": artifacts[-1].verification_details = {"risk_flags": ["pickle_or_executable_serialization"]} else: session.query(Accelerator).delete() session.commit() profile = create_profile(service, artifact_set.id) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert result.status == "blocked" def test_runtime_profile_rejects_secrets_and_is_immutable(session: Session) -> None: service, artifact_set, _node, _artifacts = setup_runtime(session) environment = create_environment(service) with pytest.raises(ValidationError, match="secrets"): RuntimeProfileCreate( name="unsafe", runtime_environment_id=environment.id, artifact_set_id=artifact_set.id, modality="embedding", environment_variables={"HF_TOKEN": "secret"}, ) safe = RuntimeProfileCreate( name="offline tokenizer", runtime_environment_id=environment.id, artifact_set_id=artifact_set.id, modality="embedding", environment_variables={"TOKENIZERS_PARALLELISM": "false"}, ) assert safe.environment_variables == {"TOKENIZERS_PARALLELISM": "false"} profile = create_profile(service, artifact_set.id) stored = service.repo.profile(profile.id) assert stored is not None stored.dtype = "float16" with pytest.raises(ValueError, match="immutable approved fields"): session.commit() session.rollback() def test_probe_requires_exact_valid_lab_approval(session: Session) -> None: service, artifact_set, node, artifacts = setup_runtime(session) profile = create_profile(service, artifact_set.id) assessment = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) other_set = ArtifactSet( revision_id=artifact_set.revision_id, snapshot_id=artifact_set.snapshot_id, variant_key="safetensors-second", label="Second set", selection_reason="approval scoping test", selected_paths=artifact_set.selected_paths, total_size_bytes=artifact_set.total_size_bytes, file_count=artifact_set.file_count, availability="local", status="verified", completeness="complete", security_status="static_checks_passed_unapproved", license_status="captured_unreviewed", immutable_at=datetime.now(UTC), ) session.add(other_set) session.flush() for ordinal, artifact in enumerate(artifacts): session.add( ArtifactSetMember( artifact_set_id=other_set.id, artifact_id=artifact.id, ordinal=ordinal, required=True, ) ) session.commit() wrong = service.approve( other_set.id, ExecutionApprovalCreate(reason="Separate reviewed artifact approval", approved_by="test"), ) with pytest.raises(RegistryConflict, match="another artifact set"): service.create_probe( RuntimeProbeCreate( compatibility_assessment_id=assessment.id, execution_approval_id=wrong.id, ) ) def test_probe_is_node_owned_cancelable_and_creates_no_candidate_early(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) _profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node) other = ComputeNode( key="workstation", hostname="workstation", enabled=True, liveness_state="online", lab_eligible=True, agent_capabilities=["runtime.probe.v1"], ) session.add(other) session.commit() assert service.claim_next(other) is None assert service.candidates() == [] cancelled = service.cancel(probe.id) assert cancelled.status == "cancelled" assert service.claim_next(node) is None def test_successful_probe_persists_measured_evidence_and_lab_candidate(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) _profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node) lease = service.claim_next(node) assert lease is not None and lease.probe_id == probe.id claimed = service.probe(probe.id) assert claimed.logs_reference == f"runtime-worker://{node.id}/{probe.id}/attempt/1" completed = service.complete( probe.id, node, AgentRuntimeProbeComplete( lease_token=lease.lease_token, load_result={"status": "passed", "load_time_ms": 1000.0}, health_result={ "process": "healthy", "runtime": "healthy", "model": "healthy", "capability": "not_routed_in_m4", }, inference_result={ "status": "passed", "shape": [1, 1024], "dimension": 1024, "finite": True, "latency_ms": 10.0, }, unload_result={"status": "passed", "reclaimed": True}, measured_resources={ "kind": "measured", "samples": {"before_load": {"used_vram_bytes": 1}}, }, runtime_facts={"offline_local_only": True, "trust_remote_code": False}, environment_fingerprint=hashlib.sha256(b"runtime").hexdigest(), ), ) assert completed.status == "completed" candidate = service.candidates()[0] assert candidate.status == "lab_ready" assert candidate.production is False assert len(service.repo.metrics(probe.id)) == 1 def test_reranking_probe_accepts_finite_ranked_score_evidence(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) profile = create_profile(service, artifact_set.id, "qwen3_reranker") assert profile.health_contract == { "process": "required", "runtime": "required", "model": "functional_ranking_required", "capability": "rag.reranking@1", } assessment = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) approval = service.approve( artifact_set.id, ExecutionApprovalCreate(reason="Reviewed exact reranker artifact set", approved_by="test"), ) probe = service.create_probe( RuntimeProbeCreate( compatibility_assessment_id=assessment.id, execution_approval_id=approval.id, ) ) lease = service.claim_next(node) assert lease is not None completed = service.complete( probe.id, node, AgentRuntimeProbeComplete( lease_token=lease.lease_token, load_result={"status": "passed", "load_time_ms": 500.0}, health_result={ "process": "healthy", "runtime": "healthy", "model": "healthy", "capability": "healthy", }, inference_result={ "status": "passed", "output_type": "ranked_scores", "count": 1, "finite": True, "score": 0.75, "latency_ms": 20.0, }, unload_result={"status": "passed", "reclaimed": True}, measured_resources={"kind": "measured", "samples": {}}, runtime_facts={"offline_local_only": True, "trust_remote_code": False}, environment_fingerprint="e" * 64, ), ) assert completed.status == "completed" assert completed.inference_result["output_type"] == "ranked_scores" assert service.candidates()[0].runtime_probe_id == probe.id def test_completion_rejects_unhealthy_or_online_dependent_evidence(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) _profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node) lease = service.claim_next(node) assert lease is not None failed = service.complete( probe.id, node, AgentRuntimeProbeComplete( lease_token=lease.lease_token, load_result={"status": "passed"}, health_result={"process": "healthy", "runtime": "healthy", "model": "unhealthy"}, inference_result={"finite": False, "dimension": 0}, unload_result={"reclaimed": True}, measured_resources={"samples": {}}, runtime_facts={"offline_local_only": True}, environment_fingerprint="f" * 64, ), ) assert failed.status == "failed" assert failed.failure_code == "HEALTHCHECK_FAILED" assert service.candidates() == [] def test_runtime_environment_and_profile_creation_are_idempotent(session: Session) -> None: service, artifact_set, _node, _artifacts = setup_runtime(session) first_environment = create_environment(service) second_environment = create_environment(service) first_profile = create_profile(service, artifact_set.id) second_profile = create_profile(service, artifact_set.id) assert first_environment.id == second_environment.id assert first_profile.id == second_profile.id assert first_profile.image_digest == first_environment.image_digest assert first_profile.version == 1 def test_runtime_environment_identity_is_immutable(session: Session) -> None: service, _artifact_set, _node, _artifacts = setup_runtime(session) environment = create_environment(service) stored = service.repo.environment(environment.id) assert stored is not None stored.image_digest = "sha256:" + "b" * 64 with pytest.raises(ValueError, match="immutable approved fields"): session.commit() session.rollback() def test_missing_runtime_dependency_blocks_static_compatibility(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) environment = service.create_environment( RuntimeEnvironmentCreate( name="missing sentence transformers", adapter="sentence_transformers", runtime_version="4.1.0", image_repository="worker", image_digest="sha256:" + "c" * 64, python_version="3.11", cuda_runtime_version="12.8", package_versions={"torch": "2.7.1", "transformers": "4.51.3"}, supported_model_types=["qwen3"], supported_formats=["safetensors"], supported_modalities=["embedding"], ) ) profile = service.create_profile( RuntimeProfileCreate( name="missing dependency", runtime_environment_id=environment.id, artifact_set_id=artifact_set.id, modality="embedding", ) ) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert "RUNTIME_DEPENDENCY_MISSING:sentence-transformers" in result.blockers def test_modality_and_compute_capability_are_enforced(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) environment = service.create_environment( RuntimeEnvironmentCreate( name="text-only sentence runtime", adapter="sentence_transformers", runtime_version="4.1.0", image_repository="worker", image_digest="sha256:" + "d" * 64, python_version="3.11", cuda_runtime_version="12.8", package_versions={"sentence-transformers": "4.1.0"}, supported_model_types=["qwen3"], supported_formats=["safetensors"], supported_modalities=["text_generation"], ) ) profile = service.create_profile( RuntimeProfileCreate( name="incompatible GPU and modality", runtime_environment_id=environment.id, artifact_set_id=artifact_set.id, modality="embedding", gpu_memory_policy={"minimum_compute_capability": "9.0"}, ) ) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert "RUNTIME_MODALITY_UNSUPPORTED" in result.blockers assert "GPU_COMPUTE_CAPABILITY_INCOMPATIBLE" in result.blockers def test_insufficient_vram_estimate_blocks_without_loading(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) service.repo.accelerators(node.id)[0].total_vram_bytes = 100 session.commit() profile = create_profile(service, artifact_set.id) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert "INSUFFICIENT_VRAM_ESTIMATE" in result.blockers assert result.evidence["resource_estimate"]["kind"] == "estimated" def test_unknown_architecture_is_requires_probe_not_claimed_compatible(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) model = service.repo.model(service.repo.revision(artifact_set.revision_id).model_id) # type: ignore[union-attr] assert model is not None model.upstream_metadata = {"pipeline_tag": "feature-extraction"} session.commit() profile = create_profile(service, artifact_set.id) result = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert result.status == "requires_probe" assert "MODEL_ARCHITECTURE_EVIDENCE_UNKNOWN" in result.warnings def test_assessment_becomes_stale_when_hardware_facts_change(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) profile = create_profile(service, artifact_set.id) assessment = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) accelerator = service.repo.accelerators(node.id)[0] accelerator.driver_version = "999.1" session.commit() refreshed = service.assessment(assessment.id) assert refreshed.stale is True assert "evidence changed" in str(refreshed.stale_reason) def test_expired_and_changed_approvals_cannot_authorize_probe(session: Session) -> None: service, artifact_set, node, artifacts = setup_runtime(session) profile = create_profile(service, artifact_set.id) assessment = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) expired = service.approve( artifact_set.id, ExecutionApprovalCreate( reason="Time bounded exact artifact review", approved_by="operator", expires_at=datetime.now(UTC) - timedelta(seconds=1), ), ) assert expired.stale is True with pytest.raises(RegistryConflict, match="valid LAB"): service.create_probe( RuntimeProbeCreate( compatibility_assessment_id=assessment.id, execution_approval_id=expired.id, ) ) current = service.approve( artifact_set.id, ExecutionApprovalCreate(reason="Second exact artifact review", approved_by="operator"), ) artifacts[0].status = "corrupt" session.commit() assert service.approval_response(service.repo.approval(current.id)).stale is True # type: ignore[arg-type] def test_disabled_node_and_wrong_artifact_location_are_blocked(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) node.enabled = False session.commit() profile = create_profile(service, artifact_set.id) disabled = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert "NODE_NOT_LAB_READY" in disabled.blockers node.enabled = True other = ComputeNode(key="other", hostname="other", enabled=True, liveness_state="online") session.add(other) session.flush() root = session.query(StorageRoot).one() root.compute_node_id = other.id session.commit() misplaced = service.assess( CompatibilityAssessmentCreate(runtime_profile_id=profile.id, compute_node_id=node.id) ) assert "VERIFIED_ARTIFACT_LOCATION_NOT_ON_TARGET_NODE" in misplaced.blockers def test_expired_worker_lease_is_recovered_after_restart(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) _profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node) first = service.claim_next(node) assert first is not None stored = service.repo.probe(probe.id) assert stored is not None stored.lease_expires_at = datetime.now(UTC) - timedelta(seconds=1) session.commit() second = service.claim_next(node) assert second is not None assert second.lease_token != first.lease_token assert service.probe(probe.id).attempt_count == 2 def test_running_probe_cancellation_is_acknowledged_and_terminal(session: Session) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) _profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node) lease = service.claim_next(node) assert lease is not None service.cancel(probe.id) control = service.progress( probe.id, node, AgentRuntimeProbeProgress(lease_token=lease.lease_token, status="loading"), ) assert control.cancel_requested is True cancelled = service.fail( probe.id, node, AgentRuntimeProbeFailure( lease_token=lease.lease_token, failure_code="CANCELLED", failure_message="operator cancelled", ), ) assert cancelled.status == "cancelled" assert service.candidates() == [] @pytest.mark.parametrize( ("health", "inference", "unload", "facts", "code"), [ ( {"process": "healthy", "runtime": "healthy", "model": "healthy"}, {"finite": False, "dimension": 0}, {"reclaimed": True}, {"offline_local_only": True}, "INVALID_OUTPUT", ), ( {"process": "healthy", "runtime": "healthy", "model": "healthy"}, {"finite": True, "dimension": 1024}, {"reclaimed": False}, {"offline_local_only": True}, "GPU_MEMORY_NOT_RECLAIMED", ), ( {"process": "healthy", "runtime": "healthy", "model": "healthy"}, {"finite": True, "dimension": 1024}, {"reclaimed": True}, {"offline_local_only": False}, "OFFLINE_LOAD_VIOLATION", ), ], ) def test_completion_failures_are_typed_and_never_create_candidate( session: Session, health: dict, inference: dict, unload: dict, facts: dict, code: str, ) -> None: service, artifact_set, node, _artifacts = setup_runtime(session) _profile, _assessment, _approval, probe = prepare_probe(service, artifact_set, node) lease = service.claim_next(node) assert lease is not None result = service.complete( probe.id, node, AgentRuntimeProbeComplete( lease_token=lease.lease_token, load_result={"status": "passed"}, health_result=health, inference_result=inference, unload_result=unload, measured_resources={"samples": {}}, runtime_facts=facts, environment_fingerprint="e" * 64, ), ) assert result.status == "failed" assert result.failure_code == code assert service.candidates() == []