SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(perception): qualify E33 worker shadow

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-27 13:43:18 +03:00
parent 32257fc2b4
commit 9bbd5845b7
13 changed files with 2448 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_e33_worker_shadow.py
+333
View File
@@ -0,0 +1,333 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
import numpy as np
import pytest
from k1link.compute.e33_worker_shadow import (
E33_PACKAGE_SCHEMA,
E33_PROFILE_SCHEMA,
E33WorkerShadowError,
read_e33_worker_shadow_result,
run_e33_worker_shadow,
)
_IMAGE = "test/e33-worker@sha256:" + "a" * 64
_AUTHORITY = {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
}
def test_e33_worker_shadow_closes_every_frame_and_reuses_existing_result(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
package = _package(tmp_path / "package-source", frame_count=8)
monkeypatch.setenv("E33_CONTAINER_IMAGE", _IMAGE)
monkeypatch.setenv("E33_WORKER_NODE", "test-worker")
result = run_e33_worker_shadow(
package,
tmp_path / "results",
execution_id="contract-pass",
)
reused = run_e33_worker_shadow(
package,
tmp_path / "results",
execution_id="contract-pass",
)
assert result.accepted is True
assert reused.result_id == result.result_id
accounting = result.report["metrics"]["accounting"]
assert accounting == {
"source_frames": 8,
"delivered": 8,
"input_superseded": 0,
"result_superseded": 0,
"closed": True,
}
assert result.report["acceptance"]["requirements"][
"navigation_or_safety_authority_false"
]
assert result.result["authority"] == _AUTHORITY
def test_e33_worker_shadow_records_overload_instead_of_hiding_loss(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
package = _package(
tmp_path / "package-source",
frame_count=24,
interval_seconds=0.005,
processing_delay_ms=30.0,
work_capacity=1,
)
monkeypatch.setenv("E33_CONTAINER_IMAGE", _IMAGE)
monkeypatch.setenv("E33_WORKER_NODE", "test-worker")
result = run_e33_worker_shadow(
package,
tmp_path / "results",
execution_id="contract-overload",
)
accounting = result.report["metrics"]["accounting"]
assert result.accepted is False
assert accounting["source_frames"] == 24
assert accounting["delivered"] + accounting["input_superseded"] == 24
assert accounting["input_superseded"] > 0
assert "input_drop_fraction_within_gate" in result.report["acceptance"][
"rejection_reasons"
]
outcomes = [
json.loads(line)
for line in (result.result_root / "frame-outcomes.jsonl").read_text().splitlines()
]
assert [row["frame_index"] for row in outcomes] == list(range(24))
assert any(row["status"] == "input-superseded" for row in outcomes)
def test_e33_reader_rejects_tampered_artifact(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
package = _package(tmp_path / "package-source", frame_count=4)
monkeypatch.setenv("E33_CONTAINER_IMAGE", _IMAGE)
monkeypatch.setenv("E33_WORKER_NODE", "test-worker")
result = run_e33_worker_shadow(
package,
tmp_path / "results",
execution_id="tamper-check",
)
with (result.result_root / "frame-outcomes.jsonl").open("a", encoding="utf-8") as stream:
stream.write("{}\n")
with pytest.raises(E33WorkerShadowError, match="artifact identity changed"):
read_e33_worker_shadow_result(result.result_root)
def test_e33_rejects_wrong_container_identity(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
package = _package(tmp_path / "package-source", frame_count=4)
monkeypatch.setenv("E33_CONTAINER_IMAGE", "different")
monkeypatch.setenv("E33_WORKER_NODE", "test-worker")
with pytest.raises(E33WorkerShadowError, match="container image identity changed"):
run_e33_worker_shadow(
package,
tmp_path / "results",
execution_id="wrong-container",
)
def test_e33_rejects_wrong_worker_node_identity(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
package = _package(tmp_path / "package-source", frame_count=4)
monkeypatch.setenv("E33_CONTAINER_IMAGE", _IMAGE)
monkeypatch.setenv("E33_WORKER_NODE", "unexpected-worker")
with pytest.raises(E33WorkerShadowError, match="worker node identity changed"):
run_e33_worker_shadow(
package,
tmp_path / "results",
execution_id="wrong-worker-node",
)
def _package(
root: Path,
*,
frame_count: int,
interval_seconds: float = 0.02,
processing_delay_ms: float = 0.0,
work_capacity: int = 2,
) -> Path:
root.mkdir(parents=True)
e32_identity = {
"frame_count": frame_count,
"timeline_start_seconds": 0.0,
"timeline_end_seconds": (frame_count - 1) * interval_seconds,
"source": {
"camera_result_id": "e10-integrated-perception-" + "1" * 64,
"source_pack_id": "e10-lidar-pack-" + "2" * 64,
},
"authority": _AUTHORITY,
}
e32_identity_sha = hashlib.sha256(_canonical(e32_identity)).hexdigest()
e32_id = f"e32-track-geometry-{e32_identity_sha}"
e32_root = root / "input" / "e32" / e32_id
e32_root.mkdir(parents=True)
offsets = np.arange(frame_count + 1, dtype="<i8")
source_indices = np.arange(frame_count, dtype="<i8")
points = np.column_stack(
(
np.arange(frame_count, dtype=np.float32),
np.zeros(frame_count, dtype=np.float32),
np.ones(frame_count, dtype=np.float32),
)
).astype("<f4")
owners = np.zeros(frame_count, dtype="<u4")
np.save(e32_root / "frame-point-offsets.npy", offsets, allow_pickle=False)
np.save(e32_root / "point-source-indices.npy", source_indices, allow_pickle=False)
np.save(e32_root / "point-coordinates-map-f32.npy", points, allow_pickle=False)
np.save(e32_root / "point-owner-indices.npy", owners, allow_pickle=False)
with (e32_root / "track-geometry-frames.jsonl").open(
"x",
encoding="utf-8",
) as stream:
for frame_index in range(frame_count):
stream.write(
json.dumps(
{
"schema_version": "missioncore.e32-track-geometry-record/v1",
"track_geometry_frame_schema": (
"missioncore.track-geometry-frame/v1"
),
"frame_index": frame_index,
"source_frame_index": 100 + frame_index,
"session_seconds": frame_index * interval_seconds,
"source_available": frame_index % 3 != 0,
"point_slab": {
"schema_version": (
"missioncore.e32-point-slab-reference/v1"
),
"contract_schema": "missioncore.point-slab/v1",
"source_point_count": 1,
"coordinate_frame": "map",
"owner_keys": ["semantic:1"],
"row_count": 1,
},
"geometries": [
{
"key": "semantic:1",
"evidence_state": "agree",
}
],
"policy": {},
"authority": _AUTHORITY,
},
sort_keys=True,
)
+ "\n"
)
e32_artifacts = [
_artifact(e32_root / name)
for name in (
"track-geometry-frames.jsonl",
"frame-point-offsets.npy",
"point-source-indices.npy",
"point-coordinates-map-f32.npy",
"point-owner-indices.npy",
)
]
_write_json(
e32_root / "manifest.json",
{
"schema_version": "missioncore.e32-track-geometry-replay/v1",
"result_id": e32_id,
"identity_sha256": e32_identity_sha,
"identity": e32_identity,
"artifacts": e32_artifacts,
"authority": _AUTHORITY,
},
)
profile = {
"schema_version": E33_PROFILE_SCHEMA,
"mode": "contract-test",
"expected_e32_result_id": e32_id,
"expected_worker_node": "test-worker",
"container_image": _IMAGE,
"pacing": {"speed": 1.0, "start_delay_ms": 10.0},
"queues": {"work_capacity": work_capacity, "result_capacity": 2},
"deadlines": {
"result_ms": 100.0,
"stale_ms": 150.0,
"unavailable_ms": 500.0,
},
"resources": {"sample_interval_seconds": 0.1},
"acceptance": {
"minimum_effective_fps": 1.0,
"maximum_input_drop_fraction": 0.0,
"maximum_result_drop_fraction": 0.0,
"maximum_result_age_p95_ms": 100.0,
"maximum_deadline_miss_fraction": 0.0,
"maximum_release_lag_p95_ms": 25.0,
"maximum_process_rss_mib": 4096.0,
"require_gpu_visible": False,
},
"test_controls": {"processing_delay_ms": processing_delay_ms},
"authority": _AUTHORITY,
}
_write_json(root / "profile.json", profile)
package_members = [
path
for path in root.rglob("*")
if path.is_file() and path.name != "manifest.json"
]
relative_paths = sorted(path.relative_to(root).as_posix() for path in package_members)
package_identity = {
"schema_version": E33_PACKAGE_SCHEMA,
"classification": "contract-test",
"e32_result_id": e32_id,
"e32_identity_sha256": e32_identity_sha,
"profile_sha256": _sha256(root / "profile.json"),
"artifact_paths": relative_paths,
"authority": _AUTHORITY,
}
package_identity_sha = hashlib.sha256(_canonical(package_identity)).hexdigest()
package_id = f"e33-worker-package-{package_identity_sha}"
manifest = {
"schema_version": E33_PACKAGE_SCHEMA,
"package_id": package_id,
"identity_sha256": package_identity_sha,
"identity": package_identity,
"artifacts": [
{
"kind": relative,
"path": relative,
"byte_length": (root / relative).stat().st_size,
"sha256": _sha256(root / relative),
}
for relative in relative_paths
],
}
_write_json(root / "manifest.json", manifest)
destination = root.parent / package_id
root.rename(destination)
return destination
def _artifact(path: Path) -> dict[str, Any]:
return {
"path": path.name,
"byte_length": path.stat().st_size,
"sha256": _sha256(path),
}
def _write_json(path: Path, value: object) -> None:
path.write_text(json.dumps(value, sort_keys=True), encoding="utf-8")
def _canonical(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
def _sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
tests/test_live_perception.py
+3 -3
View File
@@ -210,9 +210,9 @@ def test_live_result_rejects_tampering_and_partial_cuboid() -> None:
def test_latest_wins_queue_never_exceeds_capacity() -> None:
queue = LatestWinsQueue[int](capacity=2)
queue.publish(1)
queue.publish(2)
queue.publish(3)
assert queue.publish(1) is None
assert queue.publish(2) is None
assert queue.publish(3) == 1
assert queue.take_next(timeout=0) == 2
assert queue.take_next(timeout=0) == 3