tests/datasets/test_runtime_milestone8.py

from __future__ import annotations

import hashlib
import json
from pathlib import Path

import pytest
from pydantic import BaseModel, ValidationError

from gcmd_classifier.datasets.cmr import resolve_cmr_collection, validate_cmr_url
from gcmd_classifier.datasets.errors import (
    DatasetCacheError,
    DatasetOrchestrationError,
    DatasetPersistenceError,
)
from gcmd_classifier.datasets.models import (
    CMRRetrievedSource,
    CoverageMode,
    DatasetClassificationOutcome,
    DatasetClassificationResult,
    DatasetIdentity,
    DatasetProcessingMetadata,
    DatasetProcessingStatus,
    DatasetSemanticValidatorState,
    ResolvedCMRCollection,
)
from gcmd_classifier.datasets.orchestration import (
    DatasetRunController,
    DurableInferenceClient,
    recover_run,
    validate_stage_log,
)
from gcmd_classifier.datasets.pipeline import BlindDatasetPipeline
from gcmd_classifier.datasets.runtime_models import (
    ArtifactReference,
    ArtifactType,
    BlindRunState,
    CheckpointStatus,
    ModelCallAccounting,
    PipelineCheckpoint,
    PipelineStage,
    RunIdentity,
    SessionIdentity,
    StructuredStageLogEvent,
)
from gcmd_classifier.datasets.runtime_persistence import (
    DatasetArtifactStore,
    persist_expert_source_boundary,
    verify_persisted_blind_result,
)
from gcmd_classifier.datasets.stage_cache import (
    BlindClassificationCacheIdentity,
    BlindStageCache,
    CMRSourceCacheIdentity,
    READMEArtifactCacheIdentity,
    build_blind_cache_key,
)
from gcmd_classifier.llm.base import ModelRequest, ModelResponse, ModelStage
from tests.datasets.fixture_manifest import load_fixture_manifest

SHA_A = "a" * 64
SHA_B = "b" * 64


@pytest.fixture
def fixture_root() -> Path:
    return Path("tests/fixtures/datasets").resolve()


@pytest.fixture
def run() -> RunIdentity:
    return RunIdentity(run_id="run-1", concept_id="C123-TEST")


def frozen_collection(index: int = 0) -> ResolvedCMRCollection:
    _, cases = load_fixture_manifest(Path("tests/fixtures/datasets/cases.json"))
    case = cases[index]
    raw = case.cmr_path.read_bytes()
    parsed = json.loads(raw)
    source = CMRRetrievedSource(
        submitted_url=case.metadata.cmr_url,
        final_url=case.metadata.cmr_url,
        retrieved_at=case.metadata.fixture_recorded_at,
        status_code=200,
        response_metadata={"content-type": "application/json"},
        response_bytes=raw,
        source_text=raw.decode("utf-8"),
        sha256=hashlib.sha256(raw).hexdigest(),
        parsed_json=parsed,
    )
    return resolve_cmr_collection(validate_cmr_url(case.metadata.cmr_url), source)


def collection_for(
    identity: DatasetIdentity, keywords: list[dict] | None = None
) -> ResolvedCMRCollection:
    url = (
        "https://cmr.earthdata.nasa.gov/search/collections.umm_json?native_id=" + identity.native_id
    )
    parsed = {
        "hits": 1,
        "items": [
            {
                "meta": {
                    "concept-id": identity.concept_id,
                    "native-id": identity.native_id,
                    "revision-id": identity.cmr_revision_id,
                },
                "umm": {
                    "ShortName": identity.short_name,
                    "Version": identity.version,
                    "RelatedUrls": [],
                    "ScienceKeywords": keywords or [],
                },
            }
        ],
    }
    raw = json.dumps(parsed, separators=(",", ":")).encode()
    source = CMRRetrievedSource(
        submitted_url=url,
        final_url=url,
        retrieved_at="2026-08-13T00:00:00Z",
        status_code=200,
        response_bytes=raw,
        source_text=raw.decode(),
        sha256=hashlib.sha256(raw).hexdigest(),
        parsed_json=parsed,
    )
    return resolve_cmr_collection(validate_cmr_url(url), source)


def initial(run: RunIdentity) -> BlindRunState:
    return BlindRunState(
        run=run,
        session=SessionIdentity(session_id="session-1"),
        checkpoints=(
            PipelineCheckpoint(stage=PipelineStage.INITIALIZE, status=CheckpointStatus.COMPLETED),
        ),
    )


def test_storage_rejects_fixture_root(fixture_root: Path) -> None:
    with pytest.raises(DatasetPersistenceError, match="cannot be inside fixtures"):
        DatasetArtifactStore(fixture_root, fixture_root=fixture_root)


def test_artifact_store_is_atomic_immutable_and_run_isolated(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    ref = store.persist_bytes(
        run=run,
        session_id="session-1",
        relative_name="source/raw.json",
        data=b"exact",
        artifact_type=ArtifactType.CMR_RESPONSE,
    )
    assert store.load_verified(run, "session-1", ref) == b"exact"
    with pytest.raises(DatasetPersistenceError, match="overwrite"):
        store.persist_bytes(
            run=run,
            session_id="session-1",
            relative_name="source/raw.json",
            data=b"changed",
            artifact_type=ArtifactType.CMR_RESPONSE,
        )
    other = RunIdentity(run_id="run-2", concept_id=run.concept_id)
    with pytest.raises(DatasetPersistenceError):
        store.load_verified(other, "session-1", ref)


def test_atomic_failure_never_publishes_success(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    def fail(point: str) -> None:
        if point == "before_publish":
            raise RuntimeError("interrupt")

    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root, failure_hook=fail)
    with pytest.raises(DatasetPersistenceError):
        store.persist_bytes(
            run=run,
            session_id="session-1",
            relative_name="blind.json",
            data=b"{}",
            artifact_type=ArtifactType.BLIND_RESULT,
        )
    assert not (store.run_directory(run, "session-1") / "blind.json").exists()
    assert list(store.run_directory(run, "session-1").glob("*.tmp"))


def test_illegal_and_terminal_transitions_are_rejected(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    controller = DatasetRunController(
        initial(run), DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    )
    with pytest.raises(DatasetOrchestrationError):
        controller.complete(PipelineStage.CMR_SOURCE, "not begun")
    controller.begin(PipelineStage.CMR_SOURCE, "begin")
    controller.complete(PipelineStage.CMR_SOURCE, "done")
    with pytest.raises(DatasetOrchestrationError):
        controller.begin(PipelineStage.CMR_SOURCE, "stale")


class Reply(BaseModel):
    value: str


class FakeClient:
    calls = 0

    def generate_structured(self, request: ModelRequest[Reply]) -> ModelResponse[Reply]:
        self.calls += 1
        return ModelResponse(
            parsed=Reply(value="ok"), provider="fake", model_name="fake", prompt_version="v1"
        )


def request() -> ModelRequest[Reply]:
    return ModelRequest(
        stage=ModelStage.DATASET_COVERAGE,
        prompt="safe",
        response_schema=Reply,
        provider="fake",
        model_name="fake",
        prompt_version="v1",
        temperature=0,
        timeout_seconds=1,
    )


def test_inference_attempt_is_durable_before_transport(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    controller = DatasetRunController(
        initial(run), DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    )
    fake = FakeClient()
    DurableInferenceClient(fake, controller).generate_structured(request())
    assert controller.state.classification_attempted
    assert controller.state.model_calls.current_run_calls == 1
    assert fake.calls == 1


def test_failed_attempt_checkpoint_prevents_transport(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    def fail(point: str) -> None:
        if point == "before_publish":
            raise RuntimeError("disk")

    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    controller = DatasetRunController(initial(run), store)
    controller.store._failure_hook = fail
    fake = FakeClient()
    with pytest.raises(DatasetPersistenceError):
        DurableInferenceClient(fake, controller).generate_structured(request())
    assert fake.calls == 0


def test_recovery_uses_validated_checkpoint_chain(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    controller = DatasetRunController(initial(run), store)
    controller.begin(PipelineStage.CMR_SOURCE, "begin")
    controller.complete(PipelineStage.CMR_SOURCE, "done")
    recovered, decision = recover_run(store, run, "session-1")
    assert recovered == controller.state
    assert decision.value == "resume_next_stage"


def test_cache_identity_requires_complete_stage_inputs() -> None:
    first = CMRSourceCacheIdentity(
        submitted_url="https://cmr.example/a",
        final_url="https://cmr.example/a",
        response_sha256=SHA_A,
        retrieval_policy_version="v1",
    )
    changed = first.model_copy(update={"response_sha256": SHA_B})
    assert build_blind_cache_key(first) != build_blind_cache_key(changed)
    readme_a = READMEArtifactCacheIdentity(
        selected_source_entry_sha256=SHA_A,
        submitted_url="https://a.example/readme",
        final_url="https://a.example/readme",
        pdf_sha256=SHA_B,
        retrieval_policy_version="v1",
    )
    readme_b = readme_a.model_copy(update={"submitted_url": "https://b.example/readme"})
    assert build_blind_cache_key(readme_a) != build_blind_cache_key(readme_b)


def test_blind_cache_round_trip_and_tamper_detection(tmp_path: Path, fixture_root: Path) -> None:
    cache = BlindStageCache(tmp_path / "cache", fixture_root=fixture_root)
    identity = BlindClassificationCacheIdentity(
        evidence_packet_sha256=SHA_A,
        vocabulary_sha256=SHA_B,
        model_configuration_sha256="c" * 64,
        topic_prompt_version="t",
        term_prompt_version="u",
        variable_prompt_version="v",
        review_policy_version="r",
    )
    cache.put(identity, {"classification_attempted": True})
    assert cache.get(identity) == {"classification_attempted": True}
    key = build_blind_cache_key(identity)
    path = cache._store.root / "blind" / "classification" / key.identity_sha256 / "record.json"
    path.write_text("{}")
    with pytest.raises(DatasetCacheError):
        cache.get(identity)


def test_log_contract_rejects_absolute_reference(run: RunIdentity) -> None:
    event = StructuredStageLogEvent(
        timestamp="2026-01-01T00:00:00Z",
        run_id=run.run_id,
        session_id="session-1",
        stage=PipelineStage.CMR_SOURCE,
        status=CheckpointStatus.COMPLETED,
        event_code="CMR_COMPLETE",
        classification_attempted=False,
        current_run_model_calls=0,
        relative_reference="/secret/path",
    )
    with pytest.raises(DatasetOrchestrationError):
        validate_stage_log(event)


def test_verified_blind_result_gate_reads_persisted_bytes(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    identity = DatasetIdentity(
        concept_id=run.concept_id,
        native_id="SHORT_1",
        short_name="SHORT",
        version="1",
        cmr_revision_id=1,
    )
    result = DatasetClassificationResult(
        schema_version="2.0",
        dataset_key=run.concept_id,
        identity=identity,
        evidence_packet_sha256=SHA_A,
        coverage_mode=CoverageMode.SINGLE_PRODUCT,
        semantic_validator_state=DatasetSemanticValidatorState.NOT_RUN_BY_POLICY,
        classification_attempted=True,
        processing_status=DatasetProcessingStatus.COMPLETED,
        classification_outcome=DatasetClassificationOutcome.NOT_CLASSIFIED,
        no_classification_reason="No supported path.",
        processing_metadata=DatasetProcessingMetadata(model_calls=1),
    )
    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    boundary = persist_expert_source_boundary(
        store=store,
        run=run,
        session_id="session-1",
        collection=collection_for(identity, [{"Topic": "SENTINEL"}]),
    )
    ref = store.persist_json(
        run=run,
        session_id="session-1",
        relative_name="blind_result.json",
        value=result,
        artifact_type=ArtifactType.BLIND_RESULT,
    )
    verified = verify_persisted_blind_result(
        store=store,
        run=run,
        session_id="session-1",
        reference=ref,
        expected_packet_sha256=SHA_A,
        expected_native_id="SHORT_1",
        vocabulary_hash=SHA_B,
        source_boundary=boundary,
    )
    assert (
        verified.artifact.sha256
        == hashlib.sha256(store.load_verified(run, "session-1", ref)).hexdigest()
    )


def test_runtime_models_reject_unknown_fields(run: RunIdentity) -> None:
    with pytest.raises(ValidationError):
        ModelCallAccounting(current_run_calls=0, unknown=True)
    with pytest.raises(ValidationError):
        ArtifactReference(
            artifact_type="cmr_response",
            relative_path="../escape",
            sha256=SHA_A,
            byte_count=1,
            persistence_state="hash_verified",
        )


class NoReadmeServices:
    def __init__(self, collection: ResolvedCMRCollection) -> None:
        self.collection = collection
        self.stages: list[PipelineStage] = []

    def execute(self, stage: PipelineStage, context: dict, model_client: object) -> object:
        self.stages.append(stage)
        if stage is PipelineStage.CMR_SOURCE:
            return self.collection.source
        if stage is PipelineStage.IDENTITY:
            return self.collection
        if stage is PipelineStage.README_DISCOVERY:
            return None
        return {"stage": stage.value}


def test_ordered_pipeline_no_readme_is_preinference_skip(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    collection = frozen_collection()
    actual_run = RunIdentity(run_id=run.run_id, concept_id=collection.dataset_key)
    controller = DatasetRunController(initial(actual_run), store)
    services = NoReadmeServices(collection)
    terminal = BlindDatasetPipeline(
        controller=controller,
        store=store,
        services=services,
        model_client=FakeClient(),
        vocabulary_hash=SHA_A,
    ).run()
    assert terminal.terminal_outcome.value == "completed_skipped"
    assert not terminal.state.classification_attempted
    assert terminal.state.model_calls.current_run_calls == 0
    assert services.stages == [
        PipelineStage.CMR_SOURCE,
        PipelineStage.IDENTITY,
        PipelineStage.BLIND_BOUNDARY,
        PipelineStage.README_DISCOVERY,
    ]


class SuccessfulServices:
    def __init__(self, packet: object, collection: ResolvedCMRCollection) -> None:
        self.packet = packet
        self.collection = collection

    def execute(self, stage: PipelineStage, context: dict, model_client: object) -> object:
        if stage not in {PipelineStage.CMR_SOURCE, PipelineStage.IDENTITY}:
            assert "EXPERT-SENTINEL" not in str(context)
        if stage is PipelineStage.CMR_SOURCE:
            return self.collection.source
        if stage is PipelineStage.IDENTITY:
            return self.collection
        if stage is PipelineStage.PRODUCT_RESOLUTION:
            model_client.generate_structured(request())
        if stage is PipelineStage.README_RETRIEVAL:
            return b"%PDF-1.4 frozen"
        if stage is PipelineStage.EVIDENCE_PACKET:
            return self.packet
        if stage is PipelineStage.CLASSIFICATION:
            return DatasetClassificationResult(
                schema_version="2.0",
                dataset_key=self.packet.target.concept_id,
                identity=self.packet.target,
                evidence_packet_sha256=self.packet.packet_sha256,
                coverage_mode=self.packet.coverage_mode,
                semantic_validator_state=DatasetSemanticValidatorState.NOT_RUN_BY_POLICY,
                classification_attempted=True,
                processing_status=DatasetProcessingStatus.COMPLETED,
                classification_outcome=DatasetClassificationOutcome.NOT_CLASSIFIED,
                no_classification_reason="No defensible path.",
                processing_metadata=DatasetProcessingMetadata(model_calls=1),
            )
        return {"stage": stage.value}


def test_successful_not_classified_pipeline_is_persisted_and_verified(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    from tests.datasets.test_classifier_adapter import _packet

    packet = _packet(tmp_path)
    collection = collection_for(packet.target, [{"Topic": "EXPERT-SENTINEL"}])
    actual_run = RunIdentity(run_id=run.run_id, concept_id=packet.target.concept_id)
    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    controller = DatasetRunController(initial(actual_run), store)
    terminal = BlindDatasetPipeline(
        controller=controller,
        store=store,
        services=SuccessfulServices(packet, collection),
        model_client=FakeClient(),
        vocabulary_hash=SHA_A,
    ).run()
    assert terminal.terminal_outcome.value == "completed_not_classified"
    assert terminal.verified_blind_result is not None
    assert terminal.state.blind_result_hash_verified
    assert terminal.state.classification_attempted
    assert terminal.state.model_calls.current_run_calls == 1


def test_sealed_artifact_failure_stops_pipeline_before_inference(
    tmp_path: Path, fixture_root: Path, run: RunIdentity
) -> None:
    collection = frozen_collection()
    actual_run = RunIdentity(run_id=run.run_id, concept_id=collection.dataset_key)
    store = DatasetArtifactStore(tmp_path / "out", fixture_root=fixture_root)
    controller = DatasetRunController(initial(actual_run), store)
    publishes = 0

    def fail(point: str) -> None:
        nonlocal publishes
        if point == "before_publish":
            publishes += 1
            if publishes == 5:
                raise RuntimeError("sealed persistence unavailable")

    store._failure_hook = fail
    model = FakeClient()
    terminal = BlindDatasetPipeline(
        controller=controller,
        store=store,
        services=NoReadmeServices(collection),
        model_client=model,
        vocabulary_hash=SHA_A,
    ).run()
    assert terminal.terminal_outcome.value == "pre_inference_failure"
    assert not terminal.state.classification_attempted
    assert terminal.state.model_calls.current_run_calls == 0
    assert model.calls == 0