NODEDC_MISSION_CORE/tests/test_reference_graph_result.py

from __future__ import annotations

import json
from pathlib import Path

from k1link.perception.contracts import LocalObstacleMap, SourceAccounting
from k1link.perception.graph_contracts import (
    REFERENCE_GRAPH_ID_V2,
    DeliveredFrame,
    GraphRunMode,
    GraphState,
    TerminalOutcome,
    TerminalOutcomeType,
    build_graph_run_result_v2,
)
from k1link.perception.providers import ReferencePerceptionGraphConfigV2
from k1link.perception.reference_graph_identity import ReferenceGraphRuntimeIdentity
from k1link.perception.reference_graph_parity import ReferenceGraphParityReport
from k1link.perception.reference_graph_result import seal_reference_graph_result

REPOSITORY_ROOT = Path(__file__).resolve().parents[1]


def _runtime() -> ReferenceGraphRuntimeIdentity:
    return ReferenceGraphRuntimeIdentity.from_dict(
        {
            "schema_version": "missioncore.reference-graph-runtime-identity/v3",
            "worker_id": "worker-006",
            "worker_node": "DESKTOP-OPJ8J04",
            "worker_container_id": "c" * 64,
            "worker_image_id": "sha256:" + "d" * 64,
            "isolated_triton_container_id": "e" * 64,
            "isolated_triton_image_id": "sha256:" + "d" * 64,
            "historical_worker_container_id": "f" * 64,
            "historical_worker_running": False,
            "historical_triton_container_id": "a" * 64,
            "historical_triton_running": False,
            "artifact_sha256": "b" * 64,
            "patch_id": "mission-core-m47-test",
            "code_revision": "1" * 40,
            "graph_id": "reference-perception-graph/v2",
            "started_at_utc": "2026-08-23T12:00:00.000Z",
            "source_mount_read_only": True,
            "isolated_model_service": True,
            "public_worker_port_added": False,
            "commands_enabled": False,
            "actuation_allowed": False,
        }
    )


def _parity(accepted: bool = True) -> ReferenceGraphParityReport:
    return ReferenceGraphParityReport(
        expected_frames=1,
        compared_frames=1,
        temporal_frames_sha256="a" * 64,
        threat_frames_sha256="b" * 64,
        mismatch_counts=(("threat_assessments", 0 if accepted else 1),),
        accepted=accepted,
    )


def _result(outcome: TerminalOutcomeType = TerminalOutcomeType.DELIVERED):
    terminal = TerminalOutcome(
        source_id="RAVNOVES00",
        session_id="20260720T065719Z_viewer_live",
        frame_id="frame-000000",
        sequence=0,
        outcome=outcome,
        stage_id="threat" if outcome is TerminalOutcomeType.DELIVERED else "detector",
        reason=(
            "object-payload-delivered"
            if outcome is TerminalOutcomeType.DELIVERED
            else "bounded-queue-latest-wins"
        ),
    )
    deliveries = (
        (
            DeliveredFrame(
                sequence=0,
                obstacle_map=LocalObstacleMap(
                    source_id="RAVNOVES00",
                    session_id="20260720T065719Z_viewer_live",
                    frame_id="frame-000000",
                    graph_id=REFERENCE_GRAPH_ID_V2,
                    generated_monotonic_ns=123,
                    output_age_ns=45,
                    occupied=(),
                    unknown=(),
                    camera_uncertainty=(),
                    accounting=SourceAccounting(1, 1, 0, 0),
                ),
                threats=(),
            ),
        )
        if outcome is TerminalOutcomeType.DELIVERED
        else ()
    )
    return build_graph_run_result_v2(
        graph_id=REFERENCE_GRAPH_ID_V2,
        source_profile_id="m4-ravnoves00-recorded-realtime/v1",
        run_mode=GraphRunMode.LOSSLESS_REPLAY,
        state=GraphState.STOPPED,
        admitted_count=1,
        outcomes=(terminal,),
        deliveries=deliveries,
        queue_high_watermarks=(("detector", 1), ("threat", 1)),
    )


def test_m47_graph_config_round_trips_and_pins_rolling_stage() -> None:
    path = REPOSITORY_ROOT / "config/perception/m4-reference-graph-v2.json"
    document = json.loads(path.read_text("utf-8"))
    config = ReferencePerceptionGraphConfigV2.from_dict(document)

    assert config.to_dict() == document
    assert config.graph_id == REFERENCE_GRAPH_ID_V2
    assert {provider.role.value for provider in config.providers} == {
        "source",
        "detector",
        "geometry",
        "temporal",
        "motion",
        "rolling",
        "threat",
    }
    assert {queue.stage_id for queue in config.queues} == {
        "detector",
        "geometry",
        "temporal",
        "rolling",
        "threat",
    }


def test_reference_graph_result_is_content_addressed_and_reproducible(tmp_path: Path) -> None:
    first = seal_reference_graph_result(
        _result(),
        output_root=tmp_path / "one",
        expected_frames=1,
        parity=_parity(),
        runtime_identity=_runtime(),
        execution_elapsed_seconds=1.25,
    )
    second = seal_reference_graph_result(
        _result(),
        output_root=tmp_path / "two",
        expected_frames=1,
        parity=_parity(),
        runtime_identity=_runtime(),
        execution_elapsed_seconds=1.25,
    )

    assert first.accepted is True
    assert first.result_id == second.result_id
    assert first.report["gates"] == {
        "lossless_replay_mode": True,
        "graph_stopped": True,
        "admitted_frame_count": True,
        "terminal_accounting_closed": True,
        "delivery_count": True,
        "delivery_payload_count": True,
        "no_failed_frames": True,
        "no_stale_frames": True,
        "no_superseded_frames": True,
        "no_rejected_frames": True,
        "no_unavailable_frames": True,
        "accepted_m45r_m46_parity": True,
    }
    assert {path.name: path.read_bytes() for path in first.result_root.iterdir()} == {
        path.name: path.read_bytes() for path in second.result_root.iterdir()
    }
    runtime = json.loads((first.result_root / "runtime.json").read_text("utf-8"))
    assert runtime == _runtime().to_dict()
    assert first.manifest["runtime"] == runtime
    assert first.manifest["files"]["runtime.json"]["sha256"]


def test_reference_graph_result_fails_closed_on_supersession(tmp_path: Path) -> None:
    sealed = seal_reference_graph_result(
        _result(TerminalOutcomeType.SUPERSEDED),
        output_root=tmp_path,
        expected_frames=1,
        parity=_parity(),
        runtime_identity=_runtime(),
        execution_elapsed_seconds=1.25,
    )

    assert sealed.accepted is False
    assert sealed.report["gates"]["no_superseded_frames"] is False
    assert sealed.report["gates"]["delivery_count"] is False