from __future__ import annotations import stat from collections.abc import Callable from pathlib import Path from typing import Any import pytest from fastapi import APIRouter, HTTPException from fastapi.routing import APIRoute from pydantic import ValidationError from k1link.web.compute_contour_api import default_compute_contour from k1link.web.system_telemetry_api import ( EXPECTED_NODE_ID, WorkerConnectionProfile, WorkerConnectionProfilePut, WorkerProfileStore, WorkerTelemetryService, _agent_raw_document, _profile_from_compute_contour, _ssh_arguments, build_system_telemetry_router, ) def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]: for route in router.routes: if ( isinstance(route, APIRoute) and route.path == path and route.methods is not None and method in route.methods ): return route.endpoint raise AssertionError(f"{method} {path} route is missing") def _probe( *, node_id: str = EXPECTED_NODE_ID, received_bytes: float = 1_000, sent_bytes: float = 500, ) -> dict[str, Any]: return { "reachable": True, "identity_matches": node_id == EXPECTED_NODE_ID, "source": "agent-mqtt", "node_id": node_id, "latency_ms": 12.5, "observed_at_utc": "2026-07-27T12:00:00Z", "error_code": None, "raw": { "node_id": node_id, "observed_at_utc": "2026-07-27T12:00:00Z", "os": { "caption": "Windows 11 Pro", "version": "10.0", "uptime_seconds": 100, }, "cpu": { "name": "CPU", "logical_processors": 32, "load_percent": 25, }, "memory": { "total_bytes": 1000, "free_bytes": 400, }, "disks": [], "gpu": { "name": "GPU", "utilization_percent": 30, "memory_used_mib": 100, "memory_total_mib": 200, }, "network": [ { "name": "LAN", "description": "Ethernet", "status": "Up", "link_speed_bps": 1_000_000_000, "addresses": ["192.0.2.10"], "received_bytes": received_bytes, "sent_bytes": sent_bytes, } ], "docker_stats": { "mission-core-triton": { "Name": "mission-core-triton", "CPUPerc": "2.5%", "MemPerc": "3.5%", "MemUsage": "1GiB / 32GiB", "NetIO": "1MB / 2MB", "BlockIO": "0B / 0B", "PIDs": "12", }, "sentinel-frigate": { "Name": "sentinel-frigate", "CPUPerc": "150%", "MemPerc": "20%", "MemUsage": "6GiB / 32GiB", "NetIO": "3GB / 1GB", "BlockIO": "1GB / 1GB", "PIDs": "200", }, }, "container_states": { "mission-core-triton": { "state": {"Status": "running", "Health": {"Status": "healthy"}}, "image": "triton@sha256:accepted", }, "sentinel-frigate": { "state": {"Status": "running", "Health": {"Status": "healthy"}}, "image": "frigate@sha256:external", }, }, "triton": { "ready": True, "metrics": [ 'nv_inference_request_success{model="detector"} 12', 'nv_inference_request_failure{model="detector"} 0', ], }, "perception": { "state": "busy", "current_stage": "detector", "active_stages": ["detector", "semantic-model"], "active_request_id": "run-001", "active_frame_index": 42, "completed_runs": 3, "failed_runs": 0, "model_load_seconds": 10.5, "stage_metrics": { "detector": { "elapsed_seconds": 2.5, "activations": 42, "share_percent": 62.5, }, "semantic-model": { "elapsed_seconds": 1.5, "activations": 11, "share_percent": 37.5, }, }, }, }, } def test_worker_profile_rejects_ssh_option_injection() -> None: for unsafe in ( "-oProxyCommand=touch /tmp/unsafe", "worker.local -p 2200", "worker.local/../../unsafe", "worker.local\nHost evil", ): with pytest.raises(ValidationError): WorkerConnectionProfile(address=unsafe) assert WorkerConnectionProfile(address="192.0.2.15").address == "192.0.2.15" assert WorkerConnectionProfile(address="worker-006.local").address == "worker-006.local" def test_worker_profile_is_atomic_versioned_and_private(tmp_path: Path) -> None: store = WorkerProfileStore(tmp_path / "system") initial = store.read() assert initial.revision == 0 saved = store.save( WorkerConnectionProfilePut( revision=0, address="192.0.2.15", port=2200, ) ) assert saved.revision == 1 assert store.read() == saved assert stat.S_IMODE(store.path.stat().st_mode) == 0o600 with pytest.raises(RuntimeError, match="revision changed"): store.save( WorkerConnectionProfilePut( revision=0, address="192.0.2.16", port=22, ) ) def test_ssh_command_keeps_identity_pinned_and_values_as_arguments() -> None: arguments = _ssh_arguments( WorkerConnectionProfile(address="192.0.2.15", port=2200) ) assert "BatchMode=yes" in arguments assert "StrictHostKeyChecking=yes" in arguments assert "HostKeyAlias=mission-gpu" in arguments assert "HostName=192.0.2.15" in arguments assert arguments[arguments.index("-p") + 1] == "2200" assert "mission-gpu" in arguments assert "192.0.2.15; touch unsafe" not in arguments def test_worker_telemetry_separates_mission_core_and_external_load( tmp_path: Path, ) -> None: service = WorkerTelemetryService( WorkerProfileStore(tmp_path / "system"), lambda _: _probe(), cache_seconds=0, ) document = service.snapshot(10) runtimes = {runtime["name"]: runtime for runtime in document["runtimes"]} assert document["connection"]["identity_matches"] is True assert document["connection"]["source"] == "agent-mqtt" assert document["node"]["memory"]["used_percent"] == 60 assert document["node"]["gpu"]["memory_used_percent"] == 50 assert document["node"]["triton"]["requests_succeeded"] == 12 assert runtimes["mission-core-triton"]["external"] is False assert runtimes["mission-core-triton"]["cpu_percent"] == 2.5 assert runtimes["sentinel-frigate"]["external"] is True assert runtimes["sentinel-frigate"]["cpu_percent"] == 150 assert document["pipeline"]["active_request_id"] == "run-001" detector_stage = next( stage for stage in document["pipeline"]["stages"] if stage["id"] == "detector" ) assert detector_stage["state"] == "active" assert detector_stage["elapsed_seconds"] == 2.5 assert detector_stage["activations"] == 42 assert detector_stage["share_percent"] == 62.5 assert next( stage for stage in document["pipeline"]["stages"] if stage["id"] == "semantic-model" )["state"] == "active" assert next( stage for stage in document["pipeline"]["stages"] if stage["id"] == "preprocessing" )["share_percent"] is None def test_worker_telemetry_prefers_ndc_container_names_during_migration( tmp_path: Path, ) -> None: probe = _probe() raw = probe["raw"] triton_stats = raw["docker_stats"].pop("mission-core-triton") triton_stats["Name"] = "ndc-mission-core-triton" raw["docker_stats"]["ndc-mission-core-triton"] = triton_stats raw["container_states"]["ndc-mission-core-triton"] = raw[ "container_states" ].pop("mission-core-triton") service = WorkerTelemetryService( WorkerProfileStore(tmp_path / "system"), lambda _: probe, cache_seconds=0, ) document = service.snapshot(10) mission_core_runtimes = [ runtime for runtime in document["runtimes"] if not runtime["external"] ] assert len(mission_core_runtimes) == 2 triton = next( runtime for runtime in mission_core_runtimes if runtime["role"] == "Inference Runtime" ) assert triton["name"] == "ndc-mission-core-triton" assert triton["canonical_name"] == "ndc-mission-core-triton" def test_agent_pipeline_snapshot_merges_all_stage_series() -> None: samples = [] for stage_id, stage_state, elapsed, activations, share in ( ("detector", "active", 2.5, 42, 62.5), ("semantic-model", "waiting", 1.5, 11, 37.5), ): samples.append( { "node_id": EXPECTED_NODE_ID, "kind": "pipeline", "measurement": "missioncore_pipeline", "observed_at_utc": "2026-07-28T12:00:00Z", "payload": { "name": "missioncore_pipeline", "tags": {"stage_id": stage_id}, "fields": { "service_state": "busy", "current_stage": "detector", "active_request_id": "request-006", "active_frame_index": 17, "completed_runs": 3, "failed_runs": 0, "model_load_seconds": 10.5, "collector_state": "live", "stage_state": stage_state, "elapsed_seconds": elapsed, "activations": activations, "share_percent": share, }, }, } ) raw = _agent_raw_document({"samples": samples}) assert raw["perception"]["state"] == "busy" assert raw["perception"]["active_stages"] == ["detector"] assert raw["perception"]["stage_metrics"] == { "detector": { "elapsed_seconds": 2.5, "activations": 42, "share_percent": 62.5, }, "semantic-model": { "elapsed_seconds": 1.5, "activations": 11, "share_percent": 37.5, }, } def test_worker_telemetry_history_keeps_one_row_per_agent_observation( tmp_path: Path, ) -> None: service = WorkerTelemetryService( WorkerProfileStore(tmp_path / "system"), lambda _: _probe(), cache_seconds=0, ) first = service.snapshot(10) second = service.snapshot(10) assert len(first["history"]) == 1 assert len(second["history"]) == 1 assert second["history"][0]["observed_at_utc"] == "2026-07-27T12:00:00Z" def test_agent_metrics_are_mapped_to_the_existing_product_contract() -> None: document = _agent_raw_document( { "samples": [ { "node_id": EXPECTED_NODE_ID, "observed_at_utc": "2026-07-27T12:00:00Z", "kind": "host", "measurement": "cpu", "payload": { "fields": {"usage_active": 12.5}, "tags": {"cpu": "cpu-total"}, }, }, { "node_id": EXPECTED_NODE_ID, "observed_at_utc": "2026-07-27T12:00:00Z", "kind": "host", "measurement": "mem", "payload": { "fields": { "total": 1_000, "available": 400, }, "tags": {}, }, }, { "node_id": EXPECTED_NODE_ID, "observed_at_utc": "2026-07-27T12:00:00Z", "kind": "host", "measurement": "net", "payload": { "fields": { "bytes_recv": 2_000, "bytes_sent": 1_000, }, "tags": {"interface": "Ethernet"}, }, }, { "node_id": EXPECTED_NODE_ID, "observed_at_utc": "2026-07-27T12:00:00Z", "kind": "host", "measurement": "docker_container_cpu", "payload": { "fields": {"usage_percent": 2.5}, "tags": { "container_name": "ndc-mission-core-triton", "container_image": "triton@sha256:accepted", }, }, }, ] } ) assert document["node_id"] == EXPECTED_NODE_ID assert document["cpu"]["load_percent"] == 12.5 assert document["memory"] == {"total_bytes": 1_000.0, "free_bytes": 400.0} assert document["network"][0]["name"] == "Ethernet" assert document["docker_stats"]["ndc-mission-core-triton"]["CPUPerc"] == "2.500%" assert ( document["container_states"]["ndc-mission-core-triton"]["image"] == "triton@sha256:accepted" ) def test_compute_contour_maps_to_worker_identity_without_singleton_defaults() -> None: contour = default_compute_contour().model_copy( update={ "contour_id": "field-worker", "agent_id": "field-agent", "display_name": "Field Worker", "expected_node_id": "FIELD-01", "address": "192.0.2.25", } ) profile = _profile_from_compute_contour(contour) assert profile.profile_id == "field-worker" assert profile.display_name == "Field Worker" assert profile.expected_node_id == "FIELD-01" assert profile.address == "192.0.2.25" def test_profile_apply_fails_closed_on_wrong_node_and_keeps_old_profile( tmp_path: Path, ) -> None: router = build_system_telemetry_router( root_provider=lambda: tmp_path / "system", probe_runner=lambda _: _probe(node_id="OTHER-NODE"), ) apply_profile = _endpoint(router, "/api/v1/system/worker-profile", "PUT") with pytest.raises(HTTPException) as error: apply_profile( WorkerConnectionProfilePut( revision=0, address="192.0.2.20", port=22, ) ) assert error.value.status_code == 409 assert WorkerProfileStore(tmp_path / "system").read().revision == 0 assert not WorkerProfileStore(tmp_path / "system").path.exists()