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

feat(observatory): add durable recorded compute queue

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-31 01:04:25 +03:00
parent e72609120f
commit d67e86176e
23 changed files with 6300 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_observatory_m49_queue_binding.py
+555
View File
@@ -0,0 +1,555 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import cast
import pytest
import k1link.observatory.m49_queue_binding as binding_module
from k1link.artifact_gateway import (
ArtifactMember,
CentralArtifactStore,
LocalArtifactCache,
)
from k1link.observatory.m49_queue_binding import (
OBSERVATORY_EXECUTOR_BINDING_SCHEMA,
OBSERVATORY_SOURCE_BUNDLE_SCHEMA,
OBSERVATORY_SOURCE_CAPABILITY_SCHEMA,
SOURCE_DOCUMENT_STORE_DIRECTORY,
M49QueueBindingConfig,
M49QueueBindingIntegrityError,
M49RecordedQueueBindingService,
M49SourcePackIdentity,
M49SourcePackVerifier,
_verify_source_pack_from_cache,
)
from k1link.observatory.setups import LaboratorySetupRegistry
from k1link.sessions.models import (
SessionArtifact,
SessionDetail,
SessionSource,
SessionSummary,
)
from k1link.sessions.store import SessionStore
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
BINDING_PATH = (
REPOSITORY_ROOT / "config" / "observatory-m49-recorded-queue-binding.json"
)
SETUP_REGISTRY_PATH = (
REPOSITORY_ROOT / "config" / "observatory-laboratory-setups.json"
)
SOURCE_SESSION_ID = "20260720T065719Z_viewer_live"
SOURCE_LABEL = "RAVNOVES00"
SETUP_ID = "m49-tgs-full-shadow-v1"
DEFINITION_SHA256 = (
"836a66639e69f7ea00de2a9111c1c6c9c3c00f1abd726f1df596aeb4e3a88ae6"
)
SOURCE_PACK_SHA256 = (
"0685d24219d8236caf8b7f1685e93f6d6b59e7fd015a768d88a92bbe8b154944"
)
EXECUTOR_ARTIFACT_SHA256 = (
"5e0ea16c7a5cc760463836718b0cd8b0006ffc4b202e5f706a20a86ef2f912ab"
)
CODE_REVISION = "40c850b167dda366d8aa45d828520168affaf9fd"
TRAVEL_IMAGE_SHA256 = (
"7b412020f4d8392d1d1ed1b33beadc44140f0ea8f781e62dd69796042334300f"
)
PARITY_IMAGE_SHA256 = (
"ceb13548617e4bd3f619766bfdff00af3fa5160946b367828da6d2233dcdcba0"
)
SOURCE_CATALOG_SHA256 = (
"24207af81b67de515ba9f1b899577bd877a00d058c3472269508980541b1e185"
)
class _SessionStore:
def __init__(
self,
data_dir: Path,
detail: SessionDetail,
catalog_sha256: str,
) -> None:
self.data_dir = data_dir.resolve()
self.detail = detail
self.catalog_sha256 = catalog_sha256
def get_session_with_catalog_snapshot(
self, session_id: str
) -> tuple[SessionDetail, str]:
if session_id != self.detail.summary.session_id:
raise KeyError(session_id)
return self.detail, self.catalog_sha256
def _canonical_sha256(value: object) -> str:
payload = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
return hashlib.sha256(payload).hexdigest()
def test_first_immutable_contract_store_syncs_directory_chain_bottom_up(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
payload = b'{"schema_version":"test/v1"}'
digest = hashlib.sha256(payload).hexdigest()
root = tmp_path / "observatory-source-contracts"
synced: list[Path] = []
monkeypatch.setattr(
binding_module,
"_fsync_directory",
lambda path: synced.append(path),
)
binding_module._write_immutable_document(root, digest, payload)
assert synced == [
root / "objects" / "sha256" / digest[:2],
root / "objects" / "sha256",
root / "objects",
root,
tmp_path,
]
def test_existing_immutable_contract_replays_directory_chain_sync(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
payload = b'{"schema_version":"test/v1"}'
digest = hashlib.sha256(payload).hexdigest()
root = tmp_path / "observatory-source-contracts"
binding_module._write_immutable_document(root, digest, payload)
synced: list[Path] = []
monkeypatch.setattr(
binding_module,
"_fsync_directory",
lambda path: synced.append(path),
)
binding_module._write_immutable_document(root, digest, payload)
assert synced == [
root / "objects" / "sha256" / digest[:2],
root / "objects" / "sha256",
root / "objects",
root,
tmp_path,
]
def test_immutable_contract_store_rejects_intermediate_symlink(
tmp_path: Path,
) -> None:
payload = b'{"schema_version":"test/v1"}'
digest = hashlib.sha256(payload).hexdigest()
root = tmp_path / "observatory-source-contracts"
outside = tmp_path / "outside"
root.mkdir()
outside.mkdir()
(root / "objects").symlink_to(outside, target_is_directory=True)
with pytest.raises(
M49QueueBindingIntegrityError,
match="store directory is invalid",
):
binding_module._write_immutable_document(root, digest, payload)
assert list(outside.iterdir()) == []
def _detail() -> SessionDetail:
summary = SessionSummary(
session_id=SOURCE_SESSION_ID,
display_name=SOURCE_LABEL,
status="ready",
started_at_utc="2026-07-20T06:57:20.888Z",
completed_at_utc="2026-07-20T07:06:16.599Z",
duration_seconds=535.717620042,
modalities=("point-cloud", "trajectory", "video"),
source_count=3,
total_bytes=100,
replayable=True,
origin="xgrids-k1.viewer-live.evidence",
)
sources = (
SessionSource(
source_id="sensor.camera.right",
semantic_channel_id="camera.video.recorded",
modality="video",
status="recorded",
seekable=True,
artifact_id="recorded-video",
),
SessionSource(
source_id="sensor.lidar.primary",
semantic_channel_id="spatial.point-cloud.recorded",
modality="point-cloud",
status="recorded",
seekable=True,
artifact_id="raw-primary",
),
SessionSource(
source_id="spatial.trajectory",
semantic_channel_id="spatial.pose.recorded",
modality="trajectory",
status="recorded",
seekable=True,
artifact_id="raw-primary",
),
)
artifacts = (
SessionArtifact(
artifact_id="raw-primary",
kind="raw-transport",
media_type="application/x-nodedc-k1mqtt",
byte_length=64,
sha256="d" * 64,
integrity_status="verified",
),
SessionArtifact(
artifact_id="recorded-video",
kind="recorded-video",
media_type="video/mp4",
byte_length=36,
sha256=None,
integrity_status="validated-structure",
),
)
return SessionDetail(
summary=summary,
sources=sources,
artifacts=artifacts,
plugin_id="xgrids-k1",
archive_id="viewer-live",
)
def _registry() -> LaboratorySetupRegistry:
return LaboratorySetupRegistry.from_file(
SETUP_REGISTRY_PATH,
repository_root=REPOSITORY_ROOT,
)
def _artifact_cache(
data_dir: Path,
tmp_path: Path,
payload: bytes,
identity: M49SourcePackIdentity,
) -> LocalArtifactCache:
source = tmp_path / "source-pack.npz"
source.write_bytes(payload)
central = CentralArtifactStore(tmp_path / "central", create=True)
published = central.publish_file(source)
assert published.sha256 == identity.sha256
cache = LocalArtifactCache(
data_dir / "artifact-cache",
max_bytes=1024 * 1024,
free_space_reserve_bytes=0,
)
member = ArtifactMember(
role="lidar-source-pack",
media_type=identity.media_type,
sha256=published.sha256,
byte_length=published.byte_length,
)
cache.fetch(central, member, pin_id=f"session:{SOURCE_SESSION_ID}")
return cache
def _accept_source_pack(
_cache: LocalArtifactCache,
identity: M49SourcePackIdentity,
) -> None:
assert identity.sha256 == SOURCE_PACK_SHA256
assert identity.byte_length == 72_996_000
def _service(
tmp_path: Path,
*,
catalog_sha256: str = SOURCE_CATALOG_SHA256,
setup_registry: LaboratorySetupRegistry | None = None,
source_pack_verifier: M49SourcePackVerifier = _accept_source_pack,
) -> tuple[M49RecordedQueueBindingService, LocalArtifactCache]:
data_dir = tmp_path / "data"
data_dir.mkdir()
config = M49QueueBindingConfig.from_file(BINDING_PATH)
cache = LocalArtifactCache(
data_dir / "artifact-cache",
max_bytes=1024 * 1024,
free_space_reserve_bytes=0,
)
store = _SessionStore(data_dir, _detail(), catalog_sha256)
service = M49RecordedQueueBindingService(
data_dir=data_dir,
session_store=cast(SessionStore, store),
setup_registry=setup_registry or _registry(),
config=config,
artifact_cache=cache,
source_pack_verifier=source_pack_verifier,
)
return service, cache
def test_bundled_binding_freezes_the_accepted_exact_m49_identity() -> None:
config = M49QueueBindingConfig.from_file(BINDING_PATH)
definition = config.recorded_definition()
image_set = config.image_set_document()
resource = config.resource_profile.document()
assert config.source.session_id == SOURCE_SESSION_ID
assert config.source.label == SOURCE_LABEL
assert config.source.source_pack.sha256 == SOURCE_PACK_SHA256
assert config.source.source_pack.byte_length == 72_996_000
assert config.setup.setup_id == SETUP_ID
assert config.setup.definition_sha256 == DEFINITION_SHA256
assert config.executor.artifact_sha256 == EXECUTOR_ARTIFACT_SHA256
assert config.executor.code_revision == CODE_REVISION
assert config.executor.service_installed is False
assert image_set["images"] == [
{"role": "parity", "sha256": PARITY_IMAGE_SHA256},
{"role": "travel", "sha256": TRAVEL_IMAGE_SHA256},
]
assert definition.executor_image_sha256 == _canonical_sha256(image_set)
assert definition.model_release_ids == ()
assert definition.learned_models == ()
assert definition.checkpoint_policy == "non-checkpointable"
assert definition.allowed_checkpoints == ()
assert resource["restart_from_zero"] is True
assert resource["staging_discard_required"] is True
assert resource["resource_release_receipt_required"] is True
def test_check_is_read_only_and_admit_seals_path_free_source_contracts(
tmp_path: Path,
) -> None:
service, cache = _service(tmp_path)
document_root = service.data_dir / SOURCE_DOCUMENT_STORE_DIRECTORY
cache_database_mtime_ns = cache.database_path.stat().st_mtime_ns
checked = service.check(
source_session_id=SOURCE_SESSION_ID,
setup_id=SETUP_ID,
definition_sha256=DEFINITION_SHA256,
)
assert not document_root.exists()
assert cache.database_path.stat().st_mtime_ns == cache_database_mtime_ns
assert checked.source_bundle.schema_version == OBSERVATORY_SOURCE_BUNDLE_SCHEMA
assert (
checked.capability_manifest.schema_version
== OBSERVATORY_SOURCE_CAPABILITY_SCHEMA
)
assert checked.executor_binding.schema_version == OBSERVATORY_EXECUTOR_BINDING_SCHEMA
assert checked.source_catalog_sha256 == SOURCE_CATALOG_SHA256
assert checked.definition.definition_sha256 == DEFINITION_SHA256
assert checked.definition.model_release_ids == ()
assert checked.definition.checkpoint_policy == "non-checkpointable"
assert checked.source_bundle.document()["members"] == [
{
"role": "lidar-source-pack",
"artifact_id": service.config.source.source_pack.artifact_id,
"media_type": service.config.source.source_pack.media_type,
"sha256": service.config.source.source_pack.sha256,
"byte_length": service.config.source.source_pack.byte_length,
}
]
assert "path" not in json.dumps(checked.as_dict()).lower()
admitted = service.admit(
source_session_id=SOURCE_SESSION_ID,
setup_id=SETUP_ID,
definition_sha256=DEFINITION_SHA256,
)
repeated = service.admit(
source_session_id=SOURCE_SESSION_ID,
setup_id=SETUP_ID,
definition_sha256=DEFINITION_SHA256,
)
sealed_files = tuple(document_root.rglob("*.json"))
assert admitted == checked
assert repeated == admitted
assert len(sealed_files) == 3
assert {path.stem for path in sealed_files} == {
admitted.source_bundle.sha256,
admitted.capability_manifest.sha256,
admitted.executor_binding.sha256,
}
assert all(path.read_bytes() in {
admitted.source_bundle.payload,
admitted.capability_manifest.payload,
admitted.executor_binding.payload,
} for path in sealed_files)
assert service.definitions.resolve(SETUP_ID, DEFINITION_SHA256) == (
admitted.definition
)
intent = admitted.intent(idempotency_key="m49-recorded-submit-001")
assert intent.source_session_id == SOURCE_SESSION_ID
assert intent.source_catalog_sha256 == SOURCE_CATALOG_SHA256
assert intent.source_bundle_sha256 == admitted.source_bundle.sha256
assert intent.source_capability_manifest_sha256 == (
admitted.capability_manifest.sha256
)
@pytest.mark.parametrize(
("source_session_id", "setup_id", "definition_sha256", "message"),
[
("other-session", SETUP_ID, DEFINITION_SHA256, "exact source session"),
(SOURCE_SESSION_ID, "other-setup", DEFINITION_SHA256, "exact setup"),
(SOURCE_SESSION_ID, SETUP_ID, "f" * 64, "definition digest"),
],
)
def test_binding_rejects_every_other_session_setup_or_definition(
tmp_path: Path,
source_session_id: str,
setup_id: str,
definition_sha256: str,
message: str,
) -> None:
service, _cache = _service(tmp_path)
with pytest.raises(M49QueueBindingIntegrityError, match=message):
service.check(
source_session_id=source_session_id,
setup_id=setup_id,
definition_sha256=definition_sha256,
)
def test_binding_captures_the_current_session_catalog_snapshot(
tmp_path: Path,
) -> None:
service, _cache = _service(tmp_path, catalog_sha256="e" * 64)
admission = service.check(
source_session_id=SOURCE_SESSION_ID,
setup_id=SETUP_ID,
definition_sha256=DEFINITION_SHA256,
)
assert admission.source_catalog_sha256 == "e" * 64
assert admission.source_bundle.document()["catalog_sha256"] == "e" * 64
def test_binding_fails_closed_when_the_cached_source_pack_changes(
tmp_path: Path,
) -> None:
payload = b"sealed source pack fixture\n"
identity = M49SourcePackIdentity(
artifact_id="fixture-source-pack",
sha256=hashlib.sha256(payload).hexdigest(),
byte_length=len(payload),
media_type="application/vnd.nodedc.lidar-source-pack+npz",
expected_timeline_frames=2,
expected_available_lidar_frames=1,
)
data_dir = tmp_path / "data"
data_dir.mkdir()
cache = _artifact_cache(data_dir, tmp_path, payload, identity)
_verify_source_pack_from_cache(cache, identity)
cache.object_path(identity.sha256).write_bytes(b"x" * len(payload))
with pytest.raises(M49QueueBindingIntegrityError, match="artifact-cache metadata"):
_verify_source_pack_from_cache(cache, identity)
def test_binding_fails_closed_when_the_setup_registry_definition_changes(
tmp_path: Path,
) -> None:
document = json.loads(SETUP_REGISTRY_PATH.read_text(encoding="utf-8"))
document["setups"][0]["run_definition"]["definition_id"] = (
"m49-tgs-full-shadow-drift"
)
path = tmp_path / "setup-registry.json"
path.write_text(json.dumps(document), encoding="utf-8")
drifted_registry = LaboratorySetupRegistry.from_file(
path,
repository_root=REPOSITORY_ROOT,
)
service, _cache = _service(tmp_path, setup_registry=drifted_registry)
with pytest.raises(M49QueueBindingIntegrityError, match="setup definition changed"):
service.check(
source_session_id=SOURCE_SESSION_ID,
setup_id=SETUP_ID,
definition_sha256=DEFINITION_SHA256,
)
@pytest.mark.parametrize(
"mutation",
["unknown-key", "installed", "learned-model", "source-session"],
)
def test_binding_loader_rejects_authority_or_contract_expansion(
tmp_path: Path,
mutation: str,
) -> None:
document = json.loads(BINDING_PATH.read_text(encoding="utf-8"))
if mutation == "unknown-key":
document["command"] = "run arbitrary payload"
elif mutation == "installed":
document["executor"]["service_installed"] = True
elif mutation == "learned-model":
document["executor"]["learned_models"] = ["invented-model"]
else:
document["source"]["session_id"] = "different-source"
path = tmp_path / "invalid-binding.json"
path.write_text(json.dumps(document), encoding="utf-8")
with pytest.raises(M49QueueBindingIntegrityError):
M49QueueBindingConfig.from_file(path)
def test_binding_requires_the_session_store_data_dir_artifact_cache(
tmp_path: Path,
) -> None:
service, cache = _service(tmp_path)
config = service.config
foreign = LocalArtifactCache(
tmp_path / "foreign-cache",
max_bytes=1024 * 1024,
free_space_reserve_bytes=0,
)
with pytest.raises(M49QueueBindingIntegrityError, match="data-dir artifact cache"):
M49RecordedQueueBindingService(
data_dir=service.data_dir,
session_store=service._session_store, # noqa: SLF001 - exact boundary test
setup_registry=_registry(),
config=config,
artifact_cache=foreign,
)
assert cache.root == service.data_dir / "artifact-cache"
def test_binding_wraps_unavailable_default_artifact_cache(
tmp_path: Path,
) -> None:
data_dir = tmp_path / "data"
data_dir.mkdir()
(data_dir / "artifact-cache").write_text("not a directory", encoding="utf-8")
store = _SessionStore(data_dir, _detail(), SOURCE_CATALOG_SHA256)
with pytest.raises(
M49QueueBindingIntegrityError,
match="artifact cache is unavailable",
):
M49RecordedQueueBindingService(
data_dir=data_dir,
session_store=cast(SessionStore, store),
setup_registry=_registry(),
config=M49QueueBindingConfig.from_file(BINDING_PATH),
)
tests/test_observatory_recorded_jobs.py
+728
View File
@@ -0,0 +1,728 @@
from __future__ import annotations
import sqlite3
from dataclasses import replace
from pathlib import Path
import pytest
from k1link.observatory.recorded_jobs import (
LIVE_K1_PRIORITY_RANK,
RECORDED_JOB_DATABASE_NAME,
RECORDED_PRIORITY_RANK,
ObservatoryLiveLeaseIntent,
ObservatoryNonCheckpointableCancellationReceipt,
ObservatoryRecordedJobIntent,
ObservatoryRecordedJobQueue,
ObservatoryRecordedPreemptionError,
ObservatoryRecordedQueueBusyError,
ObservatoryRecordedQueueConflictError,
ObservatoryRecordedQueueIntegrityError,
ObservatoryRecordedQueueStaleClaimError,
RecordedRunDefinition,
RecordedRunDefinitionRegistry,
)
NOW = "2026-08-30T21:00:00.000Z"
DEFINITION_SHA = "a" * 64
NON_CHECKPOINTABLE_DEFINITION_SHA = "b" * 64
ADAPTER_SHA = "c" * 64
CATALOG_SHA = "d" * 64
RESULT_SHA = "e" * 64
LIVE_EPOCH_SHA = "f" * 64
SOURCE_BUNDLE_SHA = "1" * 64
SOURCE_CAPABILITIES_SHA = "2" * 64
EXECUTOR_RELEASE_SHA = "3" * 64
EXECUTOR_IMAGE_SHA = "4" * 64
MODEL_MANIFEST_SHA = "5" * 64
RESOURCE_PROFILE_SHA = "6" * 64
def _definitions() -> RecordedRunDefinitionRegistry:
return RecordedRunDefinitionRegistry(
(
RecordedRunDefinition(
setup_id="travel-tgs-eomt-v1",
definition_id="travel-tgs-eomt",
definition_version=1,
definition_sha256=DEFINITION_SHA,
source_adapter_id="sealed-session-bundle",
source_adapter_version=1,
source_adapter_sha256=ADAPTER_SHA,
executor_release_id="travel-tgs-executor-v1",
executor_release_sha256=EXECUTOR_RELEASE_SHA,
executor_image_sha256=EXECUTOR_IMAGE_SHA,
model_release_ids=("eomt-v1", "travel-tgs-v1"),
model_manifest_sha256=MODEL_MANIFEST_SHA,
resource_profile_id="worker006-single-gpu-v1",
resource_profile_sha256=RESOURCE_PROFILE_SHA,
checkpoint_policy="cooperative",
allowed_checkpoints=(
"source-bundle-ready",
"model-batch-finished",
"evidence-sealed",
),
),
RecordedRunDefinition(
setup_id="legacy-monolith-v1",
definition_id="legacy-monolith",
definition_version=1,
definition_sha256=NON_CHECKPOINTABLE_DEFINITION_SHA,
source_adapter_id="sealed-session-bundle",
source_adapter_version=1,
source_adapter_sha256=ADAPTER_SHA,
executor_release_id="legacy-monolith-executor-v1",
executor_release_sha256=EXECUTOR_RELEASE_SHA,
executor_image_sha256=EXECUTOR_IMAGE_SHA,
model_release_ids=("legacy-monolith-v1",),
model_manifest_sha256=MODEL_MANIFEST_SHA,
resource_profile_id="worker006-single-gpu-v1",
resource_profile_sha256=RESOURCE_PROFILE_SHA,
checkpoint_policy="non-checkpointable",
),
)
)
def _queue(
tmp_path: Path,
*,
with_non_checkpointable_preemptor: bool = False,
) -> ObservatoryRecordedJobQueue:
preemptor = None
if with_non_checkpointable_preemptor:
def preemptor(request):
return ObservatoryNonCheckpointableCancellationReceipt.sealed(
request,
cancellation_id=f"cancel-{request.cancellation_request_id}",
)
return ObservatoryRecordedJobQueue(
tmp_path,
definitions=_definitions(),
clock=lambda: NOW,
non_checkpointable_preemptor=preemptor,
)
def _intent(
*,
idempotency_key: str = "recorded-request-001",
source_session_id: str = "20260828T130511Z_viewer_live",
source_catalog_sha256: str = CATALOG_SHA,
setup_id: str = "travel-tgs-eomt-v1",
definition_sha256: str = DEFINITION_SHA,
) -> ObservatoryRecordedJobIntent:
return ObservatoryRecordedJobIntent(
idempotency_key=idempotency_key,
source_session_id=source_session_id,
source_catalog_sha256=source_catalog_sha256,
source_bundle_sha256=SOURCE_BUNDLE_SHA,
source_capability_manifest_sha256=SOURCE_CAPABILITIES_SHA,
setup_id=setup_id,
definition_sha256=definition_sha256,
)
def _live_intent(
*,
trigger_id: str = "k1-live-start-001",
live_session_id: str = "20260830T210000Z_k1_live",
acquisition_epoch_sha256: str = LIVE_EPOCH_SHA,
) -> ObservatoryLiveLeaseIntent:
return ObservatoryLiveLeaseIntent(
trigger_id=trigger_id,
live_session_id=live_session_id,
acquisition_epoch_sha256=acquisition_epoch_sha256,
)
def _running_job(
queue: ObservatoryRecordedJobQueue,
*,
intent: ObservatoryRecordedJobIntent | None = None,
claim_request_id: str = "worker-claim-001",
):
job, created = queue.submit(intent or _intent())
assert created is True
queue.enqueue(job.job_id)
claim = queue.claim_next(
claimant_id="recorded-worker", claim_request_id=claim_request_id
)
assert claim is not None
running = queue.start(job.job_id, claim_token=claim.claim_token)
assert running.state == "running"
return running, claim
def test_submission_is_durable_exactly_idempotent_and_path_free(tmp_path: Path) -> None:
queue = _queue(tmp_path)
first, first_created = queue.submit(_intent())
second, second_created = queue.submit(_intent())
restored = ObservatoryRecordedJobQueue(
tmp_path,
definitions=_definitions(),
clock=lambda: NOW,
).get(first.job_id)
assert first_created is True
assert second_created is False
assert second == first
assert restored == first
assert first.state == "accepted"
assert first.priority_class == "recorded"
assert first.priority_rank == RECORDED_PRIORITY_RANK
assert first.source_adapter_id == "sealed-session-bundle"
assert first.checkpoint_policy == "cooperative"
projection = first.as_dict()
assert projection["priority"] == {
"class": "recorded",
"rank": RECORDED_PRIORITY_RANK,
"server_owned": True,
}
assert "path" not in str(projection).lower()
assert not hasattr(_intent(), "command")
assert not hasattr(_intent(), "image")
assert projection["source"]["bundle_sha256"] == SOURCE_BUNDLE_SHA
assert projection["source"]["capability_manifest_sha256"] == (
SOURCE_CAPABILITIES_SHA
)
assert projection["executor"]["release_sha256"] == EXECUTOR_RELEASE_SHA
assert projection["executor"]["image_sha256"] == EXECUTOR_IMAGE_SHA
assert projection["executor"]["model_manifest_sha256"] == MODEL_MANIFEST_SHA
assert projection["executor"]["resource_profile_sha256"] == (
RESOURCE_PROFILE_SHA
)
assert queue.database_path.name == RECORDED_JOB_DATABASE_NAME
assert queue.database_path.stat().st_mode & 0o777 == 0o600
with pytest.raises(ObservatoryRecordedQueueConflictError, match="idempotency"):
queue.submit(replace(_intent(), source_session_id="another-session"))
def test_submission_and_enqueue_can_commit_as_one_idempotent_transaction(
tmp_path: Path,
) -> None:
queue = _queue(tmp_path)
first, created = queue.submit(_intent(), enqueue=True)
repeated, repeated_created = queue.submit(_intent(), enqueue=True)
assert created is True
assert repeated_created is False
assert first.state == "queued"
assert repeated == first
def test_session_setup_identity_changes_for_each_exact_combination(tmp_path: Path) -> None:
queue = _queue(tmp_path)
first, _ = queue.submit(_intent())
second, _ = queue.submit(
_intent(
idempotency_key="recorded-request-002",
source_session_id="another-session",
)
)
third, _ = queue.submit(
_intent(
idempotency_key="recorded-request-003",
setup_id="legacy-monolith-v1",
definition_sha256=NON_CHECKPOINTABLE_DEFINITION_SHA,
)
)
assert len({first.identity_sha256, second.identity_sha256, third.identity_sha256}) == 3
assert second.source_adapter_sha256 == first.source_adapter_sha256
assert third.checkpoint_policy == "non-checkpointable"
with pytest.raises(ObservatoryRecordedQueueConflictError, match="allowlisted"):
queue.submit(
_intent(
idempotency_key="recorded-request-004",
definition_sha256="0" * 64,
)
)
def test_normal_recorded_job_lifecycle_and_terminal_idempotency(tmp_path: Path) -> None:
queue = _queue(tmp_path)
running, claim = _running_job(queue)
checkpointed = queue.checkpoint(
running.job_id,
claim_token=claim.claim_token,
checkpoint_id="model-batch-finished",
)
assert checkpointed.state == "running"
assert checkpointed.last_checkpoint_id == "model-batch-finished"
succeeded = queue.succeed(
running.job_id,
claim_token=claim.claim_token,
result_id="recorded-result-001",
result_sha256=RESULT_SHA,
)
replayed = queue.succeed(
running.job_id,
claim_token=claim.claim_token,
result_id="recorded-result-001",
result_sha256=RESULT_SHA,
)
assert succeeded.state == "succeeded"
assert replayed == succeeded
assert succeeded.result_sha256 == RESULT_SHA
assert succeeded.active_claim_token is None
with pytest.raises(ObservatoryRecordedQueueConflictError, match="terminal"):
queue.fail(
running.job_id,
claim_token=claim.claim_token,
error_code="worker-failed",
message="Synthetic failure.",
)
def test_success_requires_running_and_cannot_publish_during_preemption(
tmp_path: Path,
) -> None:
queue = _queue(tmp_path)
job, _ = queue.submit(_intent())
queue.enqueue(job.job_id)
claim = queue.claim_next(
claimant_id="recorded-worker",
claim_request_id="worker-claim-001",
)
assert claim is not None
with pytest.raises(ObservatoryRecordedQueueConflictError, match="claimed"):
queue.succeed(
job.job_id,
claim_token=claim.claim_token,
result_id="recorded-result-001",
result_sha256=RESULT_SHA,
)
queue.start(job.job_id, claim_token=claim.claim_token)
queue.request_live(_live_intent())
assert queue.get(job.job_id).preemption_requested is True
with pytest.raises(
ObservatoryRecordedQueueConflictError,
match="preemption",
):
queue.succeed(
job.job_id,
claim_token=claim.claim_token,
result_id="recorded-result-001",
result_sha256=RESULT_SHA,
)
def test_claim_is_exactly_idempotent_including_empty_result(tmp_path: Path) -> None:
queue = _queue(tmp_path)
empty = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="empty-poll-001"
)
assert empty is None
job, _ = queue.submit(_intent())
queue.enqueue(job.job_id)
assert (
queue.claim_next(
claimant_id="recorded-worker", claim_request_id="empty-poll-001"
)
is None
)
claim = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="non-empty-poll-001"
)
retry = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="non-empty-poll-001"
)
assert claim is not None
assert retry is not None
assert retry.claim_token == claim.claim_token
assert retry.job.job_id == job.job_id
with pytest.raises(ObservatoryRecordedQueueConflictError, match="claim"):
queue.claim_next(
claimant_id="another-worker", claim_request_id="non-empty-poll-001"
)
def test_single_worker_resource_has_only_one_recorded_owner(tmp_path: Path) -> None:
queue = _queue(tmp_path)
first, _ = queue.submit(_intent())
second, _ = queue.submit(
_intent(
idempotency_key="recorded-request-002",
source_session_id="another-session",
)
)
queue.enqueue(first.job_id)
queue.enqueue(second.job_id)
first_claim = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="owner-poll-001"
)
assert first_claim is not None
assert (
queue.claim_next(
claimant_id="recorded-worker", claim_request_id="owner-poll-002"
)
is None
)
completed_job_id = first_claim.job.job_id
queue.fail(
completed_job_id,
claim_token=first_claim.claim_token,
error_code="synthetic-failure",
message="Release the synthetic single-Worker ownership.",
)
second_claim = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="owner-poll-003"
)
assert second_claim is not None
assert second_claim.job.job_id in {first.job_id, second.job_id} - {completed_job_id}
def test_live_lease_cooperatively_pauses_and_resumes_recorded_job(tmp_path: Path) -> None:
queue = _queue(tmp_path)
running, claim = _running_job(queue)
lease, created = queue.request_live(_live_intent())
exact_retry, retry_created = queue.request_live(_live_intent())
requested_job = queue.get(running.job_id)
assert created is True
assert retry_created is False
assert exact_retry == lease
assert lease.state == "pending"
assert lease.priority_rank == LIVE_K1_PRIORITY_RANK
assert requested_job.state == "running"
assert requested_job.preemption_requested is True
assert queue.admission_gate().blocked is True
with pytest.raises(ObservatoryRecordedQueueBusyError, match="safe preemption"):
queue.activate_live(lease.lease_id)
paused = queue.checkpoint(
running.job_id,
claim_token=claim.claim_token,
checkpoint_id="model-batch-finished",
)
assert paused.state == "paused"
active = queue.activate_live(lease.lease_id)
assert active.state == "active"
another, _ = queue.submit(
_intent(
idempotency_key="recorded-request-002",
source_session_id="another-session",
)
)
queue.enqueue(another.job_id)
assert (
queue.claim_next(
claimant_id="recorded-worker", claim_request_id="blocked-poll-001"
)
is None
)
completed = queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-finish-001",
outcome="completed",
)
completed_retry = queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-finish-001",
outcome="completed",
)
assert completed.state == "completed"
assert completed_retry == completed
assert queue.admission_gate().blocked is False
resumed = queue.get(running.job_id)
assert resumed.state == "queued"
assert resumed.preemption_requested is False
assert resumed.active_claim_token is None
with pytest.raises(ObservatoryRecordedQueueConflictError, match="terminal"):
queue.finish_live(
lease.lease_id,
terminal_trigger_id="different-terminal-trigger",
outcome="failed",
)
def test_cancelled_pending_live_lease_releases_cooperative_preemption_request(
tmp_path: Path,
) -> None:
queue = _queue(tmp_path)
running, claim = _running_job(queue)
lease, _ = queue.request_live(_live_intent())
assert queue.get(running.job_id).preemption_requested is True
cancelled = queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-cancel-001",
outcome="cancelled",
)
released = queue.get(running.job_id)
assert cancelled.state == "cancelled"
assert released.state == "running"
assert released.preemption_requested is False
assert released.active_claim_token == claim.claim_token
def test_live_request_pauses_claimed_job_before_execution(tmp_path: Path) -> None:
queue = _queue(tmp_path)
job, _ = queue.submit(_intent())
queue.enqueue(job.job_id)
claim = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="worker-claim-001"
)
assert claim is not None
lease, _ = queue.request_live(_live_intent())
paused = queue.get(job.job_id)
assert paused.state == "paused"
assert queue.start(job.job_id, claim_token=claim.claim_token).state == "paused"
assert queue.activate_live(lease.lease_id).state == "active"
queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-finish-001",
outcome="completed",
)
with pytest.raises(ObservatoryRecordedQueueStaleClaimError, match="stale"):
queue.start(job.job_id, claim_token=claim.claim_token)
def test_non_checkpointable_job_is_cancelled_and_restarts_from_zero_for_live(
tmp_path: Path,
) -> None:
queue = _queue(tmp_path, with_non_checkpointable_preemptor=True)
running, claim = _running_job(
queue,
intent=_intent(
setup_id="legacy-monolith-v1",
definition_sha256=NON_CHECKPOINTABLE_DEFINITION_SHA,
),
)
lease, _ = queue.request_live(_live_intent())
paused = queue.get(running.job_id)
assert paused.state == "paused"
assert paused.preemption_requested is True
assert paused.restart_from_zero is True
assert paused.preemption_receipt_sha256 is not None
assert paused.active_claim_token is None
with pytest.raises(ObservatoryRecordedQueueStaleClaimError, match="stale"):
queue.checkpoint(
running.job_id,
claim_token=claim.claim_token,
checkpoint_id="model-batch-finished",
)
assert queue.activate_live(lease.lease_id).state == "active"
assert (
queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-finish-001",
outcome="failed",
).state
== "failed"
)
resumed = queue.get(running.job_id)
assert resumed.state == "queued"
assert resumed.restart_from_zero is True
new_claim = queue.claim_next(
claimant_id="recorded-worker", claim_request_id="worker-claim-002"
)
assert new_claim is not None
assert new_claim.job.job_id == running.job_id
assert new_claim.job.restart_from_zero is True
def test_live_request_fails_closed_without_non_checkpointable_canceler(
tmp_path: Path,
) -> None:
queue = _queue(tmp_path)
running, _claim = _running_job(
queue,
intent=_intent(
setup_id="legacy-monolith-v1",
definition_sha256=NON_CHECKPOINTABLE_DEFINITION_SHA,
),
)
with pytest.raises(ObservatoryRecordedPreemptionError, match="scheduler-owned"):
queue.request_live(_live_intent())
assert queue.admission_gate().blocked is True
assert queue.get(running.job_id).state == "preemption-pending"
with sqlite3.connect(queue.database_path) as connection:
intent_state = connection.execute(
"SELECT state FROM observatory_recorded_preemptions"
).fetchone()[0]
assert intent_state == "pending"
recovered = _queue(tmp_path, with_non_checkpointable_preemptor=True)
lease, created = recovered.request_live(_live_intent())
assert created is False
assert recovered.get(running.job_id).state == "paused"
assert recovered.get(running.job_id).restart_from_zero is True
assert recovered.activate_live(lease.lease_id).state == "active"
def test_durable_cancel_intent_survives_callback_crash_and_reuses_identity(
tmp_path: Path,
) -> None:
cancellation_request_ids: list[str] = []
def crash_after_external_side_effect(request):
cancellation_request_ids.append(request.cancellation_request_id)
raise OSError("synthetic scheduler process loss after cancellation")
queue = ObservatoryRecordedJobQueue(
tmp_path,
definitions=_definitions(),
clock=lambda: NOW,
non_checkpointable_preemptor=crash_after_external_side_effect,
)
running, _claim = _running_job(
queue,
intent=_intent(
setup_id="legacy-monolith-v1",
definition_sha256=NON_CHECKPOINTABLE_DEFINITION_SHA,
),
)
with pytest.raises(ObservatoryRecordedPreemptionError, match="did not release"):
queue.request_live(_live_intent())
assert queue.get(running.job_id).state == "preemption-pending"
def replay_exact_cancel(request):
cancellation_request_ids.append(request.cancellation_request_id)
return ObservatoryNonCheckpointableCancellationReceipt.sealed(
request,
cancellation_id=f"cancel-{request.cancellation_request_id}",
)
recovered = ObservatoryRecordedJobQueue(
tmp_path,
definitions=_definitions(),
clock=lambda: NOW,
non_checkpointable_preemptor=replay_exact_cancel,
)
lease, created = recovered.request_live(_live_intent())
assert created is False
assert len(cancellation_request_ids) == 2
assert cancellation_request_ids[0] == cancellation_request_ids[1]
assert recovered.get(running.job_id).state == "paused"
assert recovered.activate_live(lease.lease_id).state == "active"
def test_learned_model_release_list_may_be_empty_for_algorithm_only_tgs() -> None:
definition = RecordedRunDefinition(
setup_id="travel-tgs-algorithm-v1",
definition_id="travel-tgs-algorithm",
definition_version=1,
definition_sha256="7" * 64,
source_adapter_id="sealed-session-bundle",
source_adapter_version=1,
source_adapter_sha256=ADAPTER_SHA,
executor_release_id="travel-tgs-executor-v1",
executor_release_sha256=EXECUTOR_RELEASE_SHA,
executor_image_sha256=EXECUTOR_IMAGE_SHA,
model_release_ids=(),
model_manifest_sha256=MODEL_MANIFEST_SHA,
resource_profile_id="worker006-single-gpu-v1",
resource_profile_sha256=RESOURCE_PROFILE_SHA,
checkpoint_policy="cooperative",
allowed_checkpoints=("model-batch-finished",),
)
assert definition.model_release_ids == ()
assert definition.learned_models == ()
assert definition.model_manifest_sha256 == MODEL_MANIFEST_SHA
def test_reconciliation_required_is_durable_terminal_state(tmp_path: Path) -> None:
queue = _queue(tmp_path)
running, claim = _running_job(queue)
uncertain = queue.require_reconciliation(
running.job_id,
claim_token=claim.claim_token,
reason_code="worker-outcome-unknown",
message="Dispatch was accepted but its terminal receipt is unavailable.",
)
restored = ObservatoryRecordedJobQueue(
tmp_path,
definitions=_definitions(),
clock=lambda: NOW,
).get(running.job_id)
assert uncertain.state == "reconciliation-required"
assert restored == uncertain
def test_reconciliation_required_quarantines_recorded_and_live_ownership(
tmp_path: Path,
) -> None:
queue = _queue(tmp_path)
running, claim = _running_job(queue)
queue.require_reconciliation(
running.job_id,
claim_token=claim.claim_token,
reason_code="worker-outcome-unknown",
message="Worker ownership cannot be proven released.",
)
waiting, _ = queue.submit(
_intent(idempotency_key="recorded-request-002")
)
queue.enqueue(waiting.job_id)
assert queue.claim_next(
claimant_id="recorded-worker",
claim_request_id="worker-claim-after-reconciliation",
) is None
lease, _ = queue.request_live(_live_intent())
with pytest.raises(ObservatoryRecordedQueueBusyError, match="recorded work"):
queue.activate_live(lease.lease_id)
def test_live_lease_requires_explicit_valid_terminal_transition(tmp_path: Path) -> None:
queue = _queue(tmp_path)
lease, _ = queue.request_live(_live_intent())
with pytest.raises(ObservatoryRecordedQueueConflictError, match="complete"):
queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-finish-001",
outcome="completed",
)
cancelled = queue.finish_live(
lease.lease_id,
terminal_trigger_id="k1-live-cancel-001",
outcome="cancelled",
)
assert cancelled.state == "cancelled"
assert queue.admission_gate().blocked is False
def test_queue_detects_mutated_immutable_identity(tmp_path: Path) -> None:
queue = _queue(tmp_path)
job, _ = queue.submit(_intent())
with sqlite3.connect(queue.database_path) as connection:
connection.execute(
"UPDATE observatory_recorded_jobs SET source_catalog_sha256 = ? "
"WHERE job_id = ?",
("0" * 64, job.job_id),
)
connection.commit()
with pytest.raises(ObservatoryRecordedQueueIntegrityError, match="identity"):
queue.get(job.job_id)
tests/test_observatory_recorded_run_api.py
+253
View File
@@ -0,0 +1,253 @@
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast
from fastapi import FastAPI
from fastapi.testclient import TestClient
from k1link.observatory.m49_queue_binding import M49QueueBindingIntegrityError
from k1link.observatory.recorded_jobs import (
ObservatoryRecordedJobIntent,
ObservatoryRecordedJobQueue,
RecordedRunDefinition,
RecordedRunDefinitionRegistry,
)
from k1link.observatory.setups import LaboratorySetupRegistry
from k1link.sessions import SessionNotFoundError
from k1link.sessions.models import SessionSummary
from k1link.web.observatory_api import build_observatory_router
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
REGISTRY_PATH = REPOSITORY_ROOT / "config" / "observatory-laboratory-setups.json"
RAV00_SESSION_ID = "20260720T065719Z_viewer_live"
SETUP_ID = "m49-tgs-full-shadow-v1"
CATALOG_SHA256 = "1" * 64
SOURCE_BUNDLE_SHA256 = "2" * 64
CAPABILITY_SHA256 = "3" * 64
def _source() -> SessionSummary:
return SessionSummary(
session_id=RAV00_SESSION_ID,
display_name="RAVNOVES00",
status="ready",
started_at_utc="2026-07-20T06:57:19Z",
completed_at_utc="2026-07-20T07:06:15Z",
duration_seconds=536.0,
modalities=("point-cloud", "trajectory", "video"),
source_count=3,
total_bytes=1,
replayable=True,
origin="recorded",
)
class _Store:
def get_session(self, session_id: str) -> SimpleNamespace:
if session_id != RAV00_SESSION_ID:
raise SessionNotFoundError(session_id)
return SimpleNamespace(summary=_source())
class _Admission:
def __init__(
self,
definition: RecordedRunDefinition,
catalog_sha256: str,
) -> None:
self.definition = definition
self.catalog_sha256 = catalog_sha256
def intent(self, *, idempotency_key: str) -> ObservatoryRecordedJobIntent:
return ObservatoryRecordedJobIntent(
idempotency_key=idempotency_key,
source_session_id=RAV00_SESSION_ID,
source_catalog_sha256=self.catalog_sha256,
source_bundle_sha256=SOURCE_BUNDLE_SHA256,
source_capability_manifest_sha256=CAPABILITY_SHA256,
setup_id=self.definition.setup_id,
definition_sha256=self.definition.definition_sha256,
)
class _BindingService:
def __init__(self, definition: RecordedRunDefinition) -> None:
self.config = SimpleNamespace(setup=SimpleNamespace(setup_id=SETUP_ID))
self.definition = definition
self.catalog_sha256 = CATALOG_SHA256
self.check_count = 0
self.admit_count = 0
def check(
self,
*,
source_session_id: str,
setup_id: str,
definition_sha256: str,
) -> _Admission:
self.check_count += 1
return self._resolve(source_session_id, setup_id, definition_sha256)
def admit(
self,
*,
source_session_id: str,
setup_id: str,
definition_sha256: str,
) -> _Admission:
self.admit_count += 1
return self._resolve(source_session_id, setup_id, definition_sha256)
def _resolve(
self,
source_session_id: str,
setup_id: str,
definition_sha256: str,
) -> _Admission:
if (
source_session_id != RAV00_SESSION_ID
or setup_id != self.definition.setup_id
or definition_sha256 != self.definition.definition_sha256
):
raise M49QueueBindingIntegrityError("not the exact M4.9 binding")
return _Admission(self.definition, self.catalog_sha256)
def _services(
tmp_path: Path,
) -> tuple[
LaboratorySetupRegistry,
_BindingService,
ObservatoryRecordedJobQueue,
RecordedRunDefinition,
]:
registry = LaboratorySetupRegistry.from_file(
REGISTRY_PATH,
repository_root=REPOSITORY_ROOT,
)
catalog = registry.catalog(_source())
setups = cast(list[dict[str, Any]], catalog["setups"])
definition_document = cast(dict[str, object], setups[0]["run_definition"])
definition = RecordedRunDefinition(
setup_id=SETUP_ID,
definition_id=str(definition_document["definition_id"]),
definition_version=int(definition_document["version"]),
definition_sha256=str(definition_document["definition_sha256"]),
source_adapter_id="ravnoves00-m49-source-pack",
source_adapter_version=1,
source_adapter_sha256="4" * 64,
executor_release_id="m49-tgs-full-shadow-worker-release",
executor_release_sha256="5" * 64,
executor_image_sha256="6" * 64,
model_release_ids=(),
model_manifest_sha256="7" * 64,
resource_profile_id="worker006-cpu-single-run-v1",
resource_profile_sha256="8" * 64,
checkpoint_policy="non-checkpointable",
)
binding = _BindingService(definition)
queue = ObservatoryRecordedJobQueue(
tmp_path,
definitions=RecordedRunDefinitionRegistry((definition,)),
)
return registry, binding, queue, definition
def test_exact_m49_preflight_is_queueable_and_submit_is_idempotently_queued(
tmp_path: Path,
) -> None:
registry, binding, queue, definition = _services(tmp_path)
app = FastAPI()
app.include_router(
build_observatory_router(
_Store(), # type: ignore[arg-type]
setup_registry=registry,
recorded_binding_service=binding, # type: ignore[arg-type]
recorded_job_queue=queue,
)
)
client = TestClient(app)
preflight = client.post(
"/api/v1/observatory/run-preflights",
json={
"schema_version": "missioncore.observatory-run-preflight-request/v1",
"source_session_id": RAV00_SESSION_ID,
"setup_id": SETUP_ID,
"definition_sha256": definition.definition_sha256,
},
)
assert preflight.status_code == 200
assert preflight.json()["outcome"] == "queueable"
assert preflight.json()["submission_allowed"] is True
assert next(
check
for check in preflight.json()["checks"]
if check["check_id"] == "durable-queue"
)["outcome"] == "pass"
assert binding.check_count == 1
assert binding.admit_count == 0
request = {
"schema_version": "missioncore.observatory-recorded-run-submit/v1",
"idempotency_key": "observatory-ui:m49:stable-request",
"source_session_id": RAV00_SESSION_ID,
"setup_id": SETUP_ID,
}
submitted = client.post("/api/v1/observatory/runs", json=request)
binding.catalog_sha256 = "9" * 64
repeated = client.post("/api/v1/observatory/runs", json=request)
assert submitted.status_code == 202
assert repeated.status_code == 202
assert submitted.json()["job_id"] == repeated.json()["job_id"]
assert submitted.json()["state"] == "queued"
assert submitted.json()["source"]["session_id"] == RAV00_SESSION_ID
assert submitted.json()["setup"]["setup_id"] == SETUP_ID
assert submitted.json()["executor"]["learned_models"] == []
assert submitted.json()["priority"] == {
"class": "recorded",
"rank": 100,
"server_owned": True,
}
assert binding.admit_count == 1
listed = client.get(
"/api/v1/observatory/runs",
params={"source_session_id": RAV00_SESSION_ID, "setup_id": SETUP_ID},
)
assert listed.status_code == 200
assert listed.json()["schema_version"] == (
"missioncore.observatory-recorded-job-list/v1"
)
assert [item["job_id"] for item in listed.json()["items"]] == [
submitted.json()["job_id"]
]
assert queue.admission_gate().blocked is False
def test_recorded_run_routes_fail_closed_when_queue_initialization_failed() -> None:
registry = LaboratorySetupRegistry.from_file(
REGISTRY_PATH,
repository_root=REPOSITORY_ROOT,
)
app = FastAPI()
app.include_router(
build_observatory_router(
_Store(), # type: ignore[arg-type]
setup_registry=registry,
recorded_job_queue_error="queue database is incompatible",
)
)
client = TestClient(app)
response = client.get(
"/api/v1/observatory/runs",
params={"source_session_id": RAV00_SESSION_ID, "setup_id": SETUP_ID},
)
assert response.status_code == 503
assert response.json()["detail"] == "Durable-очередь расчётов недоступна."
tests/test_observatory_setups.py
+3
View File
@@ -127,6 +127,7 @@ def test_repository_setup_registry_keeps_real_definition_and_pre_definition_resu
]
rav00 = registry.catalog(_source(RAV00_SESSION_ID, "RAVNOVES00"))
m49, current = rav00["setups"]
assert m49["display_name"] == "M4.9T5 · TRAVEL TGS · CPU-only, без ML"
assert m49["origin"] == "archived-definition"
assert m49["compatibility"]["compatible"] is True
assert m49["preflight"] == {
@@ -153,6 +154,8 @@ def test_repository_setup_registry_keeps_real_definition_and_pre_definition_resu
available_observatory_result_ids=frozenset({RAV004_RESULT_ID}),
)
existing = rav004["setups"][1]
assert existing["display_name"] == "LAB V1 · EoMT Cityscapes Large 1024 + DDRNet-39"
assert "полный TGS и независимый YOLOX отсутствуют" in existing["description"]
assert existing["origin"] == "existing-result"
assert existing["run_definition"] is None
assert existing["preflight"]["outcome"] == "existing"