diff --git a/experiments/perception/worker/observatory_portable/migrate_claim_transport_v3.py b/experiments/perception/worker/observatory_portable/migrate_claim_transport_v3.py new file mode 100644 index 0000000..a6066fd --- /dev/null +++ b/experiments/perception/worker/observatory_portable/migrate_claim_transport_v3.py @@ -0,0 +1,396 @@ +"""Bounded Worker006 control-agent migration; never alters a compute package. + +Run inside an existing agent image with Docker's socket and a new /migration +evidence directory mounted. Plan is read-only. Apply requires its exact hash. +It commits a one-file, offline child layer, preserves the executor declarations, +and retains stopped predecessors (restart=no) for explicit rollback. The saved +create-body is the durable transport declaration; old package launchers must +not replace it with the v2 parent image. No model/executor is started here. +""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import http.client +import io +import json +import socket +import tarfile +import time +from datetime import UTC, datetime +from pathlib import Path +from urllib.parse import quote, urlencode + +INSTALLED_PARENT_SHA = "5ad7d95baac63af13812cb693d492add4e806a333aba8e60edb2ea1aba754373" +M49_PARENT_SHA = "d00274a2a76e254d79b28f887b421cbc267a0483fca0e83e8c2e1c416ac8593c" +TARGETS = { + "ndc-observatory-installed-lab-worker-agent": INSTALLED_PARENT_SHA, + "ndc-observatory-m49-worker-agent": M49_PARENT_SHA, +} +SOURCE = "/opt/nodedc/mission-core/src/k1link/observatory/worker_http_transport.py" +BEFORE_SHA = "fb67ae174c8da66f7d04022310be663b2735fff72966883aaa5b50fdbc3e89c7" +OLD = b"missioncore.observatory-worker-claim-request/v2" +NEW = b"missioncore.observatory-worker-claim-request/v3" +PARENT_LABEL = "com.nodedc.claim-transport-parent.sha256" + + +def canonical(value: object) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + + +def sha(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def patch_source(payload: bytes) -> bytes: + if sha(payload) != BEFORE_SHA or payload.count(OLD) != 1: + raise ValueError("installed transport source is not the reviewed baseline") + result = payload.replace(OLD, NEW) + compile(result, SOURCE, "exec") + return result + + +class UnixConnection(http.client.HTTPConnection): + def connect(self) -> None: + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(45) + self.sock.connect("/var/run/docker.sock") + + +class Engine: + def request(self, method: str, path: str, body=None, *, raw=False): + connection = UnixConnection("localhost", timeout=45) + payload = body if isinstance(body, bytes) else (None if body is None else canonical(body)) + content_type = "application/x-tar" if isinstance(body, bytes) else "application/json" + try: + connection.request(method, "/v1.45" + path, payload, {"Content-Type": content_type}) + response = connection.getresponse() + data = response.read(2 * 1024 * 1024 + 1) + if not 200 <= response.status < 300 or len(data) > 2 * 1024 * 1024: + raise RuntimeError(f"Docker request failed: {method} {path}: {response.status}") + return data if raw else (json.loads(data) if data else None) + finally: + connection.close() + + def inspect(self, name: str): + return self.request("GET", f"/containers/{quote(name, safe='')}/json") + + def execute_json(self, name: str, source: str): + execution = self.request( + "POST", + f"/containers/{name}/exec", + { + "AttachStdout": True, + "AttachStderr": True, + "Cmd": ["python3", "-c", source], + "Env": ["PYTHONDONTWRITEBYTECODE=1"], + }, + )["Id"] + data = self.request("POST", f"/exec/{execution}/start", {}, raw=True) + status = self.request("GET", f"/exec/{execution}/json") + if status["Running"] or status["ExitCode"] != 0: + raise RuntimeError("control-only readiness/probe failed; no credential output retained") + streams = bytearray() + while data: + if len(data) < 8: + raise RuntimeError("truncated Docker exec stream") + size = int.from_bytes(data[4:8], "big") + if data[0] != 1 or size > len(data) - 8: + raise RuntimeError("unexpected Docker exec output") + streams.extend(data[8 : 8 + size]) + data = data[8 + size :] + return json.loads(streams) + + +READINESS = """import json, os, pathlib, httpx +headers = {"Authorization": "Bearer " + pathlib.Path(os.environ[ + "MISSIONCORE_OBSERVATORY_WORKER_TOKEN_FILE"]).read_text().strip(), + "X-Mission-Core-Contour-Id": "worker-006"} +with httpx.Client(base_url=os.environ["MISSIONCORE_OBSERVATORY_WORKER_BASE_URL"], + headers=headers, timeout=10, trust_env=False) as client: + response = client.get("/api/v1/worker/observatory/recorded-jobs/claim-readiness") + assert response.status_code == 200 + print(json.dumps(response.json())) +""" + +PROBE = """import json, os, pathlib, uuid +from k1link.observatory.worker_http_transport import ObservatoryWorkerHttpGateway +with ObservatoryWorkerHttpGateway( + base_url=os.environ["MISSIONCORE_OBSERVATORY_WORKER_BASE_URL"], + bearer_token=pathlib.Path(os.environ["MISSIONCORE_OBSERVATORY_WORKER_TOKEN_FILE"]).read_text().strip(), + work_root=pathlib.Path(os.environ["MISSIONCORE_OBSERVATORY_WORKER_WORK_ROOT"]), +) as client: + for index in range(8): + assert client.claim_next(claimant_id="worker-006", + claim_request_id="v3-control-probe-" + uuid.uuid4().hex, + supported_executor_identities=()) is None +print(json.dumps({"idle_claims": 8, "capabilities": [], "model_jobs": 0})) +""" + + +def require_idle(readiness: dict) -> None: + if readiness.get("schema_version") != "missioncore.observatory-claim-readiness/v1": + raise ValueError("backend v3 readiness is unavailable") + if readiness["open_live_lease_count"] or any( + count + for state, count in readiness["recorded_jobs_by_state"].items() + if state not in {"failed", "succeeded", "cancelled"} + ): + raise ValueError("queue is not idle; migration must not interrupt work") + if not readiness["protocols"][1]["grant_capacity_available"]: + raise ValueError("v3 grant capacity is unavailable") + + +def validate_target(name: str, inspection: dict) -> None: + config, host = inspection["Config"], inspection["HostConfig"] + if inspection["Name"] != "/" + name or inspection["Image"] != "sha256:" + TARGETS[name]: + raise ValueError("target container or parent image changed") + if not inspection["State"]["Running"] or not host["ReadonlyRootfs"]: + raise ValueError("expected durable read-only agent is not running") + if host["NetworkMode"] != "bridge" or host.get("DeviceRequests") or host.get("Privileged"): + raise ValueError("unexpected network/GPU/privileged agent configuration") + labels = config["Labels"] + if ( + labels.get("com.nodedc.product") != "mission-core" + or labels.get("com.nodedc.authority") != "observation-only" + ): + raise ValueError("agent ownership or authority changed") + for entry in config["Env"]: + key = entry.split("=", 1)[0].upper() + sensitive = any(word in key for word in ("TOKEN", "PASSWORD", "SECRET")) + if sensitive and not key.endswith("_FILE"): + raise ValueError("inline credential is not allowed in a transport declaration") + + +def plan(engine: Engine) -> dict: + targets = [] + for name, parent in TARGETS.items(): + inspection = engine.inspect(name) + validate_target(name, inspection) + readiness = engine.execute_json(name, READINESS) + require_idle(readiness) + targets.append( + { + "name": name, + "container_id": inspection["Id"], + "parent_image_sha256": parent, + "create_body_sha256": sha( + canonical( + { + "Config": inspection["Config"], + "HostConfig": inspection["HostConfig"], + } + ) + ), + } + ) + return { + "schema_version": "missioncore.claim-transport-migration-plan/v1", + "targets": targets, + "before_source_sha256": BEFORE_SHA, + "script_sha256": sha(Path(__file__).read_bytes()), + "model_jobs": 0, + } + + +def save(path: Path, value: object) -> None: + # New immutable evidence only. An existing plan/release is never overwritten. + with path.open("xb") as output: + output.write(canonical(value) + b"\n") + + +def replacement_body(inspection: dict, image: dict) -> dict: + body = copy.deepcopy(inspection["Config"]) + body["Image"] = image["image"] + body["Labels"].update( + { + PARENT_LABEL: image["parent_image_sha256"], + "com.nodedc.claim-protocol": "3", + "com.nodedc.claim-transport-source.sha256": image["source_sha256"], + } + ) + body["HostConfig"] = copy.deepcopy(inspection["HostConfig"]) + return body + + +def build_transport(engine: Engine, name: str, parent: str, script_sha: str) -> dict: + temporary = engine.request( + "POST", + "/containers/create?" + + urlencode( + { + "name": name + "-claim-v3-layer", + } + ), + {"Image": "sha256:" + parent, "HostConfig": {"NetworkMode": "none"}}, + )["Id"] + try: + archive = engine.request( + "GET", f"/containers/{temporary}/archive?" + urlencode({"path": SOURCE}), raw=True + ) + with tarfile.open(fileobj=io.BytesIO(archive)) as source_tar: + members = source_tar.getmembers() + if len(members) != 1 or not members[0].isfile() or members[0].size > 512_000: + raise ValueError("unexpected transport source archive") + member = members[0] + original = source_tar.extractfile(member).read() + replacement = patch_source(original) + member.name = Path(SOURCE).name + member.size = len(replacement) + member.pax_headers = {} + data = io.BytesIO() + with tarfile.open(fileobj=data, mode="w") as output: + output.addfile(member, io.BytesIO(replacement)) + engine.request( + "PUT", + f"/containers/{temporary}/archive?" + + urlencode( + { + "path": str(Path(SOURCE).parent), + } + ), + data.getvalue(), + ) + changes = engine.request("GET", f"/containers/{temporary}/changes") + allowed = {str(path) for path in Path(SOURCE).parents} | {SOURCE} + if not changes or any(item["Kind"] != 0 or item["Path"] not in allowed for item in changes): + raise ValueError("transport layer changed files outside the reviewed source") + original_image = engine.request("GET", f"/images/sha256:{parent}/json") + config = copy.deepcopy(original_image["Config"]) + config.setdefault("Labels", {}).update( + { + PARENT_LABEL: parent, + "com.nodedc.claim-protocol": "3", + "com.nodedc.claim-transport-source.sha256": sha(replacement), + "com.nodedc.claim-migration-script.sha256": script_sha, + } + ) + image = engine.request( + "POST", + "/commit?" + + urlencode( + { + "container": temporary, + "repo": name + "-claim-transport", + "tag": "v3", + } + ), + config, + )["Id"] + actual = engine.request("GET", f"/images/{image}/json") + if actual["RootFS"]["Layers"][:-1] != original_image["RootFS"]["Layers"]: + raise ValueError("transport image did not retain the exact parent layers") + return { + "image": image, + "parent_image_sha256": parent, + "source_sha256": sha(replacement), + "filesystem_changes": changes, + } + finally: + engine.request("DELETE", f"/containers/{temporary}") + + +def apply(engine: Engine, expected_plan_sha: str, evidence: Path) -> dict: + started = time.monotonic_ns() + proposal = plan(engine) + if sha(canonical(proposal)) != expected_plan_sha: + raise ValueError("migration plan changed") + evidence.mkdir(parents=False, exist_ok=False) + save(evidence / "plan.json", proposal) + results = [] + for target in proposal["targets"]: + name = target["name"] + inspection = engine.inspect(name) + validate_target(name, inspection) + create_sha = sha( + canonical( + { + "Config": inspection["Config"], + "HostConfig": inspection["HostConfig"], + } + ) + ) + if inspection["Id"] != target["container_id"] or create_sha != target["create_body_sha256"]: + raise ValueError("agent changed after plan") + require_idle(engine.execute_json(name, READINESS)) + image = build_transport( + engine, name, target["parent_image_sha256"], proposal["script_sha256"] + ) + body = replacement_body(inspection, image) + # Reusing the old hostname is intentional; paths, resources, argv and env stay exact. + backup = name + "-v2-rollback-" + inspection["Id"][:12] + declaration = { + "name": name, + "create_body": body, + "rollback_name": backup, + "rollback_container_id": inspection["Id"], + "rollback_restart_policy": inspection["HostConfig"]["RestartPolicy"], + "transport": image, + } + save(evidence / (name + "-declaration.json"), declaration) + require_idle(engine.execute_json(name, READINESS)) + engine.request("POST", f"/containers/{inspection['Id']}/stop?t=15") + engine.request( + "POST", f"/containers/{inspection['Id']}/update", {"RestartPolicy": {"Name": "no"}} + ) + engine.request( + "POST", f"/containers/{inspection['Id']}/rename?" + urlencode({"name": backup}) + ) + created = engine.request("POST", "/containers/create?" + urlencode({"name": name}), body)[ + "Id" + ] + engine.request("POST", f"/containers/{created}/start") + # Never automatically delete/roll back a started agent: new operator work may arrive. + # On failure the exact predecessor and create declaration remain for reconciliation. + time.sleep(3) + after = engine.inspect(name) + if not after["State"]["Running"] or after["RestartCount"] != 0: + raise RuntimeError("replacement agent needs reconciliation; predecessor retained") + before_probe = engine.execute_json(name, READINESS) + require_idle(before_probe) + probe = engine.execute_json(name, PROBE) + after_probe = engine.execute_json(name, READINESS) + if before_probe != after_probe: + raise RuntimeError("queue state changed during non-binding control probe") + result = { + "name": name, + "container_id": created, + "transport": image, + "probe": probe, + "readiness": after_probe, + "rollback_name": backup, + } + save(evidence / (name + "-acceptance.json"), result) + results.append(result) + receipt = { + "schema_version": "missioncore.claim-transport-migration/v1", + "utc": datetime.now(UTC).isoformat(), + "monotonic_start_ns": started, + "monotonic_end_ns": time.monotonic_ns(), + "plan_sha256": expected_plan_sha, + "agents": results, + "compute_packages_changed": False, + "model_jobs": 0, + } + save(evidence / "receipt.json", receipt) + return receipt + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--apply-plan-sha256") + parser.add_argument("--evidence", type=Path, default=Path("/migration/release")) + arguments = parser.parse_args() + engine = Engine() + if arguments.apply_plan_sha256: + result = apply(engine, arguments.apply_plan_sha256, arguments.evidence) + else: + result = plan(engine) + result = {"plan": result, "plan_sha256": sha(canonical(result))} + print(json.dumps(result, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/src/k1link/observatory/recorded_jobs.py b/src/k1link/observatory/recorded_jobs.py index dfbc3fd..2e00dea 100644 --- a/src/k1link/observatory/recorded_jobs.py +++ b/src/k1link/observatory/recorded_jobs.py @@ -44,6 +44,9 @@ OBSERVATORY_LIVE_LEASE_REQUEST_SCHEMA: Final = "missioncore.observatory-live-k1- RECORDED_JOB_DATABASE_NAME: Final = "observatory-recorded-jobs.sqlite3" MAX_RECORDED_JOBS: Final = 10_000 MAX_RECORDED_CLAIM_RECEIPTS: Final = 50_000 +# The legacy ledger is immutable history, including successful empty polls. +# v3 has a separate bounded ledger containing only actual ownership grants. +MAX_RECORDED_V3_CLAIM_RECEIPTS: Final = 50_000 # Every capability expands to four SQLite bind parameters. Keeping the public # bound at 128 stays comfortably below SQLite's traditional 999-variable limit # even when the runtime was compiled with conservative defaults. @@ -166,6 +169,16 @@ CREATE TABLE IF NOT EXISTS observatory_recorded_claim_receipts ( FOREIGN KEY (job_id) REFERENCES observatory_recorded_jobs(job_id) ); +CREATE TABLE IF NOT EXISTS observatory_recorded_claim_grants_v3 ( + claim_request_id TEXT PRIMARY KEY, + request_sha256 TEXT NOT NULL, + claimant_id TEXT NOT NULL, + job_id TEXT NOT NULL, + claim_token TEXT NOT NULL, + created_at_utc TEXT NOT NULL, + FOREIGN KEY (job_id) REFERENCES observatory_recorded_jobs(job_id) +); + CREATE TABLE IF NOT EXISTS observatory_recorded_reconciliations ( reconciliation_id TEXT PRIMARY KEY, request_sha256 TEXT NOT NULL, @@ -1152,11 +1165,13 @@ class ObservatoryRecordedJobQueue: non_checkpointable_preemptor: (ObservatoryNonCheckpointablePreemptor | None) = None, max_jobs: int = MAX_RECORDED_JOBS, max_claim_receipts: int = MAX_RECORDED_CLAIM_RECEIPTS, + max_v3_claim_receipts: int = MAX_RECORDED_V3_CLAIM_RECEIPTS, max_live_leases: int = MAX_LIVE_LEASES, claim_lease_seconds: int = DEFAULT_RECORDED_CLAIM_LEASE_SECONDS, ) -> None: _validate_quota(max_jobs, MAX_RECORDED_JOBS, "recorded job") _validate_quota(max_claim_receipts, MAX_RECORDED_CLAIM_RECEIPTS, "claim receipt") + _validate_quota(max_v3_claim_receipts, MAX_RECORDED_V3_CLAIM_RECEIPTS, "v3 claim grant") _validate_quota(max_live_leases, MAX_LIVE_LEASES, "live lease") _validate_claim_lease_seconds(claim_lease_seconds) self.data_dir = data_dir.expanduser().resolve() @@ -1166,6 +1181,7 @@ class ObservatoryRecordedJobQueue: self._non_checkpointable_preemptor = non_checkpointable_preemptor self._max_jobs = max_jobs self._max_claim_receipts = max_claim_receipts + self._max_v3_claim_receipts = max_v3_claim_receipts self._max_live_leases = max_live_leases self._claim_lease_seconds = claim_lease_seconds self._lock = threading.RLock() @@ -1313,20 +1329,69 @@ class ObservatoryRecordedJobQueue: return job raise ObservatoryRecordedQueueConflictError("job cannot be enqueued") + def claim_readiness(self) -> dict[str, object]: + """Report finite grant capacity separately from the legacy idle ledger.""" + + with self._read_connection() as connection: + legacy_count = connection.execute( + "SELECT COUNT(*) FROM observatory_recorded_claim_receipts" + ).fetchone()[0] + grant_count = connection.execute( + "SELECT COUNT(*) FROM observatory_recorded_claim_grants_v3" + ).fetchone()[0] + job_counts = dict(connection.execute( + "SELECT state, COUNT(*) FROM observatory_recorded_jobs GROUP BY state" + ).fetchall()) + live_count = connection.execute( + "SELECT COUNT(*) FROM observatory_live_leases " + "WHERE state IN ('pending', 'active')" + ).fetchone()[0] + return { + "schema_version": "missioncore.observatory-claim-readiness/v1", + "protocols": [ + { + "version": 2, + "idle_policy": "durable-receipt", + "receipt_count": legacy_count, + "receipt_limit": self._max_claim_receipts, + "grant_capacity_available": legacy_count < self._max_claim_receipts, + }, + { + "version": 3, + "idle_policy": "non-binding-no-receipt", + "receipt_count": grant_count, + "receipt_limit": self._max_v3_claim_receipts, + "grant_capacity_available": grant_count < self._max_v3_claim_receipts, + }, + ], + "storage_limit_bytes": MAX_RECORDED_JOB_STORAGE_BYTES, + "recorded_jobs_by_state": job_counts, + "open_live_lease_count": live_count, + "authority": dict(_AUTHORITY), + } + def claim_next( self, *, claimant_id: str, claim_request_id: str, supported_executor_identities: tuple[RecordedExecutorIdentity, ...] | None = None, + protocol_version: Literal[2, 3] = 2, ) -> ObservatoryRecordedClaim | None: - """Claim one compatible job atomically; even an empty claim is idempotent. + """Claim compatible work atomically with a versioned idle-poll policy. ``None`` preserves the legacy v1 claim semantics during rollout. A concrete tuple is the capability-aware v2 contract; an empty tuple - deliberately claims nothing. + deliberately claims nothing. v2 records even idle polls. v3 idle is a + non-binding observation; only a real grant is persisted. A retry of a + granted request always returns the same grant (or a stale-claim error). + Both versions honor every previously persisted receipt. """ + if isinstance(protocol_version, bool) or protocol_version not in (2, 3): + raise ValueError("claim protocol version is invalid") + if protocol_version == 3 and supported_executor_identities is None: + raise ValueError("v3 claims require an explicit capability snapshot") _validate_pattern(claimant_id, _IDENTIFIER, "claimant id") _validate_pattern(claim_request_id, _IDEMPOTENCY_KEY, "claim request id") capabilities = _canonical_executor_capabilities(supported_executor_identities) @@ -1338,22 +1403,33 @@ class ObservatoryRecordedJobQueue: with self._transaction() as connection: now = self._timestamp() self._recover_stale_claims(connection, now=now) - receipt = connection.execute( - "SELECT * FROM observatory_recorded_claim_receipts WHERE claim_request_id = ?", - (claim_request_id,), - ).fetchone() + receipts = connection.execute( + "SELECT * FROM observatory_recorded_claim_receipts WHERE claim_request_id = ? " + "UNION ALL SELECT * FROM observatory_recorded_claim_grants_v3 " + "WHERE claim_request_id = ?", + (claim_request_id, claim_request_id), + ).fetchall() + if len(receipts) > 1: + raise ObservatoryRecordedQueueIntegrityError("claim id exists in both ledgers") + receipt = receipts[0] if receipts else None if receipt is not None: if receipt["request_sha256"] != request_sha256: raise ObservatoryRecordedQueueConflictError( "claim request id is bound to another claimant" ) return self._claim_from_receipt(connection, receipt) - self._require_capacity( - connection, - table="observatory_recorded_claim_receipts", - limit=self._max_claim_receipts, - label="claim receipt", + receipt_table = ( + "observatory_recorded_claim_grants_v3" + if protocol_version == 3 + else "observatory_recorded_claim_receipts" ) + if protocol_version == 2: + self._require_capacity( + connection, + table=receipt_table, + limit=self._max_claim_receipts, + label="legacy claim receipt", + ) row = None if self._open_live_lease_row(connection) is None: active_owner = connection.execute( @@ -1368,6 +1444,8 @@ class ObservatoryRecordedJobQueue: supported_executor_identities=capabilities, ) if row is None: + if protocol_version == 3: + return None connection.execute( "INSERT INTO observatory_recorded_claim_receipts " "(claim_request_id, request_sha256, claimant_id, created_at_utc) " @@ -1375,6 +1453,13 @@ class ObservatoryRecordedJobQueue: (claim_request_id, request_sha256, claimant_id, now), ) return None + if protocol_version == 3: + self._require_capacity( + connection, + table=receipt_table, + limit=self._max_v3_claim_receipts, + label="v3 claim grant", + ) job_id = str(row["job_id"]) claim_token = hashlib.sha256( f"{uuid4().hex}:{job_id}:{claim_request_id}".encode() @@ -1405,7 +1490,7 @@ class ObservatoryRecordedJobQueue: "recorded-job claim lost its serialized state" ) connection.execute( - "INSERT INTO observatory_recorded_claim_receipts " + f"INSERT INTO {receipt_table} " "(claim_request_id, request_sha256, claimant_id, job_id, " "claim_token, created_at_utc) VALUES (?, ?, ?, ?, ?, ?)", ( @@ -2679,6 +2764,7 @@ class ObservatoryRecordedJobQueue: expected = { "observatory_recorded_jobs": 50, "observatory_recorded_claim_receipts": 6, + "observatory_recorded_claim_grants_v3": 6, "observatory_recorded_reconciliations": 18, "observatory_live_leases": 13, "observatory_recorded_preemptions": 14, @@ -2819,6 +2905,11 @@ class ObservatoryRecordedJobQueue: self._max_claim_receipts, "claim receipt", ), + ( + "observatory_recorded_claim_grants_v3", + self._max_v3_claim_receipts, + "v3 claim grant", + ), ("observatory_live_leases", self._max_live_leases, "live lease"), ( "observatory_recorded_preemptions", diff --git a/src/k1link/observatory/worker_http_transport.py b/src/k1link/observatory/worker_http_transport.py index 7e9d254..5a4509e 100644 --- a/src/k1link/observatory/worker_http_transport.py +++ b/src/k1link/observatory/worker_http_transport.py @@ -213,7 +213,7 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): "POST", "/api/v1/worker/observatory/recorded-jobs/claims", json_body={ - "schema_version": "missioncore.observatory-worker-claim-request/v2", + "schema_version": "missioncore.observatory-worker-claim-request/v3", "claim_request_id": claim_request_id, "supported_executor_identities": [ identity.as_dict() diff --git a/src/k1link/web/observatory_worker_api.py b/src/k1link/web/observatory_worker_api.py index 17e8380..7aeb0fe 100644 --- a/src/k1link/web/observatory_worker_api.py +++ b/src/k1link/web/observatory_worker_api.py @@ -158,7 +158,10 @@ class ObservatoryWorkerExecutorCapability(_StrictWorkerRequest): class ObservatoryWorkerClaimRequest(_StrictWorkerRequest): - schema_version: Literal["missioncore.observatory-worker-claim-request/v2"] + schema_version: Literal[ + "missioncore.observatory-worker-claim-request/v2", + "missioncore.observatory-worker-claim-request/v3", + ] claim_request_id: str = Field( min_length=1, max_length=160, @@ -276,6 +279,10 @@ def build_observatory_worker_router( dependencies=[Depends(require_configured_worker)], ) + @router.get("/recorded-jobs/claim-readiness") + def claim_readiness() -> dict[str, object]: + return _queue_call(queue.claim_readiness) + @router.post("/recorded-jobs/claims", response_model=None) def claim_next( request: ObservatoryWorkerClaimRequest, @@ -288,10 +295,15 @@ def build_observatory_worker_router( capability.recorded_identity() for capability in request.supported_executor_identities ), + protocol_version=(3 if request.schema_version.endswith("/v3") else 2), ) ) if claim is None: - return Response(status_code=204) + return Response(status_code=204, headers={ + "X-Mission-Core-Idle-Claim": ( + "non-binding" if request.schema_version.endswith("/v3") else "durable" + ), + }) return claim.as_dict() @router.get("/recorded-jobs/{job_id}") diff --git a/tests/test_observatory_claim_transport_migration.py b/tests/test_observatory_claim_transport_migration.py new file mode 100644 index 0000000..ce18a3e --- /dev/null +++ b/tests/test_observatory_claim_transport_migration.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +from pathlib import Path + +import pytest + +SCRIPT = ( + Path(__file__).resolve().parents[1] / "experiments/perception/worker/" + "observatory_portable/migrate_claim_transport_v3.py" +) +SPEC = importlib.util.spec_from_file_location("claim_migration", SCRIPT) +assert SPEC and SPEC.loader +migration = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(migration) + + +def test_patch_is_exact_and_does_not_upgrade_other_transport_code(monkeypatch) -> None: + source = b'PROTOCOL = "missioncore.observatory-worker-claim-request/v2"\n' + monkeypatch.setattr(migration, "BEFORE_SHA", migration.sha(source)) + assert migration.patch_source(source) == source.replace(migration.OLD, migration.NEW) + with pytest.raises(ValueError): + migration.patch_source(source + b"# unreviewed change\n") + with pytest.raises(ValueError): + migration.patch_source(source.replace(migration.OLD, migration.NEW)) + + +@pytest.mark.parametrize( + "state", ["queued", "accepted", "claimed", "running", "paused", "reconciliation-required"] +) +def test_migration_refuses_pending_or_active_work(state: str) -> None: + readiness = { + "schema_version": "missioncore.observatory-claim-readiness/v1", + "open_live_lease_count": 0, + "recorded_jobs_by_state": {state: 1}, + "protocols": [{}, {"grant_capacity_available": True}], + } + with pytest.raises(ValueError, match="queue is not idle"): + migration.require_idle(readiness) + readiness["recorded_jobs_by_state"] = {"failed": 10, "succeeded": 1} + migration.require_idle(readiness) + readiness["open_live_lease_count"] = 1 + with pytest.raises(ValueError, match="queue is not idle"): + migration.require_idle(readiness) + + +def test_target_validation_is_pinned_and_refuses_gpu_or_inline_credentials() -> None: + name, image = next(iter(migration.TARGETS.items())) + inspection = { + "Name": "/" + name, + "Image": "sha256:" + image, + "State": {"Running": True}, + "Config": { + "Labels": { + "com.nodedc.product": "mission-core", + "com.nodedc.authority": "observation-only", + }, + "Env": ["MISSIONCORE_OBSERVATORY_WORKER_TOKEN_FILE=/run/secrets/token"], + }, + "HostConfig": {"ReadonlyRootfs": True, "NetworkMode": "bridge"}, + } + migration.validate_target(name, inspection) + inspection["HostConfig"]["DeviceRequests"] = [{"Count": -1}] + with pytest.raises(ValueError): + migration.validate_target(name, inspection) + inspection["HostConfig"]["DeviceRequests"] = [] + inspection["Config"]["Env"] = ["BEARER_TOKEN=synthetic-fixture"] + with pytest.raises(ValueError, match="inline credential"): + migration.validate_target(name, inspection) + + +def test_evidence_is_immutable_and_control_probe_cannot_claim_models(tmp_path: Path) -> None: + target = tmp_path / "receipt.json" + migration.save(target, {"state": "original"}) + with pytest.raises(FileExistsError): + migration.save(target, {"state": "overwritten"}) + assert json.loads(target.read_bytes()) == {"state": "original"} + assert "supported_executor_identities=()" in migration.PROBE + assert "range(8)" in migration.PROBE + compile(migration.PROBE, "control-probe", "exec") + compile(migration.READINESS, "readiness", "exec") + + +def test_transport_clone_retains_executor_configuration_and_rollback_snapshot() -> None: + inspection = { + "Config": { + "Image": "old", + "Cmd": ["-m", "existing-worker"], + "Labels": {"com.nodedc.package-sha256": "unchanged"}, + "Env": ["DEFINITIONS=/release/definitions.json"], + "User": "0:0", + }, + "HostConfig": { + "Mounts": [{"Source": "immutable-release", "Target": "/release"}], + "RestartPolicy": {"Name": "unless-stopped"}, + "DeviceRequests": [], + }, + } + before = copy.deepcopy(inspection) + body = migration.replacement_body( + inspection, + { + "image": "new", + "parent_image_sha256": "old", + "source_sha256": "patched-source", + }, + ) + assert inspection == before + assert body["HostConfig"] == before["HostConfig"] + assert body["Labels"]["com.nodedc.package-sha256"] == "unchanged" + for key in ("Cmd", "Env", "User"): + assert body[key] == before["Config"][key] + assert body["Image"] == "new" + body["HostConfig"]["RestartPolicy"]["Name"] = "no" + assert inspection == before + + +def test_changed_plan_cannot_write_evidence_or_touch_containers( + tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setattr(migration, "plan", lambda engine: {"state": "changed"}) + evidence = tmp_path / "not-created" + with pytest.raises(ValueError, match="plan changed"): + migration.apply(object(), "0" * 64, evidence) + assert not evidence.exists() diff --git a/tests/test_observatory_recorded_jobs.py b/tests/test_observatory_recorded_jobs.py index 34f9ce6..7dee2f9 100644 --- a/tests/test_observatory_recorded_jobs.py +++ b/tests/test_observatory_recorded_jobs.py @@ -18,6 +18,7 @@ from k1link.observatory.recorded_jobs import ( ObservatoryRecordedJobQueue, ObservatoryRecordedPreemptionError, ObservatoryRecordedQueueBusyError, + ObservatoryRecordedQueueCapacityError, ObservatoryRecordedQueueConflictError, ObservatoryRecordedQueueDuplicateError, ObservatoryRecordedQueueIntegrityError, @@ -355,6 +356,144 @@ def test_duplicate_guard_keeps_new_version_of_same_setup(tmp_path: Path) -> None assert created and len(reopened.list_jobs()) == 2 +def _capabilities() -> tuple[RecordedExecutorIdentity, ...]: + definition = _definitions().definitions[0] + return (RecordedExecutorIdentity( + release_sha256=definition.executor_release_sha256, + image_sha256=definition.executor_image_sha256, + model_manifest_sha256=definition.model_manifest_sha256, + resource_profile_sha256=definition.resource_profile_sha256, + ),) + + +def _claim_rows(queue: ObservatoryRecordedJobQueue, table: str) -> list[tuple]: + assert table in {"observatory_recorded_claim_receipts", "observatory_recorded_claim_grants_v3"} + with sqlite3.connect(queue.database_path) as connection: + return connection.execute(f"SELECT * FROM {table} ORDER BY claim_request_id").fetchall() + + +def test_v3_idle_does_not_consume_grant_quota_or_modify_legacy_receipts(tmp_path: Path) -> None: + queue = ObservatoryRecordedJobQueue( + tmp_path, definitions=_definitions(), clock=lambda: NOW, + max_claim_receipts=1, max_v3_claim_receipts=1, + ) + args = {"claimant_id": "recorded-worker", "supported_executor_identities": _capabilities()} + assert queue.claim_next(**args, claim_request_id="legacy-empty") is None + legacy = _claim_rows(queue, "observatory_recorded_claim_receipts") + readiness = queue.claim_readiness() + assert readiness["protocols"][0]["grant_capacity_available"] is False + assert readiness["protocols"][1]["grant_capacity_available"] is True + with pytest.raises(ObservatoryRecordedQueueCapacityError): + queue.claim_next(**args, claim_request_id="legacy-full") + for number in range(8): + assert queue.claim_next( + **args, claim_request_id=f"idle-v3-{number}", protocol_version=3, + ) is None + assert _claim_rows(queue, "observatory_recorded_claim_grants_v3") == [] + assert _claim_rows(queue, "observatory_recorded_claim_receipts") == legacy + + job, _ = queue.submit(_intent(), enqueue=True) + # Old empty receipts stay empty even when retried via the new protocol. + assert queue.claim_next(**args, claim_request_id="legacy-empty", protocol_version=3) is None + claim = queue.claim_next(**args, claim_request_id="idle-v3-0", protocol_version=3) + assert claim is not None and claim.job.job_id == job.job_id + assert len(_claim_rows(queue, "observatory_recorded_claim_grants_v3")) == 1 + assert _claim_rows(queue, "observatory_recorded_claim_receipts") == legacy + reopened = ObservatoryRecordedJobQueue( + tmp_path, definitions=_definitions(), clock=lambda: NOW, + max_claim_receipts=1, max_v3_claim_receipts=1, + ) + assert reopened.claim_next(**args, claim_request_id="idle-v3-0", protocol_version=3) == claim + # No second grant even if a client retries a v3 grant through legacy v2. + assert reopened.claim_next(**args, claim_request_id="idle-v3-0") == claim + with pytest.raises(ObservatoryRecordedQueueConflictError): + reopened.claim_next( + claimant_id="other-worker", supported_executor_identities=_capabilities(), + claim_request_id="idle-v3-0", protocol_version=3, + ) + assert reopened.claim_next(**args, claim_request_id="busy-v3", protocol_version=3) is None + + reopened.start(job.job_id, claim_token=claim.claim_token) + reopened.succeed( + job.job_id, claim_token=claim.claim_token, + result_id="result-first", result_sha256=RESULT_SHA, + ) + with pytest.raises(ObservatoryRecordedQueueStaleClaimError): + reopened.claim_next(**args, claim_request_id="idle-v3-0", protocol_version=3) + assert reopened.claim_next(**args, claim_request_id="empty-full-v3", protocol_version=3) is None + second, _ = reopened.submit(_intent(idempotency_key="second-job"), enqueue=True) + with pytest.raises(ObservatoryRecordedQueueCapacityError): + reopened.claim_next(**args, claim_request_id="grant-full-v3", protocol_version=3) + assert reopened.get(second.job_id).state == "queued" + assert reopened.get(second.job_id).claim_generation == 0 + assert reopened.claim_readiness()["protocols"][1]["grant_capacity_available"] is False + + +def test_v3_empty_capabilities_and_live_lease_cannot_claim_work(tmp_path: Path) -> None: + queue = _queue(tmp_path) + job, _ = queue.submit(_intent(), enqueue=True) + assert queue.claim_next( + claimant_id="recorded-worker", claim_request_id="no-capabilities", + supported_executor_identities=(), protocol_version=3, + ) is None + queue.request_live(_live_intent()) + assert queue.claim_next( + claimant_id="recorded-worker", claim_request_id="live-has-priority", + supported_executor_identities=_capabilities(), protocol_version=3, + ) is None + assert queue.get(job.job_id).state == "queued" + assert _claim_rows(queue, "observatory_recorded_claim_grants_v3") == [] + + +def test_v3_two_simultaneous_retries_receive_one_grant(tmp_path: Path) -> None: + queues = (_queue(tmp_path), _queue(tmp_path)) + queues[0].submit(_intent(), enqueue=True) + barrier = Barrier(2) + + def claim(index: int): + barrier.wait(timeout=5) + return queues[index].claim_next( + claimant_id="recorded-worker", claim_request_id="same-v3-id", + supported_executor_identities=_capabilities(), protocol_version=3, + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + claims = tuple(executor.map(claim, (0, 1))) + assert claims[0] is not None and claims[0] == claims[1] + assert len(_claim_rows(queues[0], "observatory_recorded_claim_grants_v3")) == 1 + + +def test_v3_expired_grant_cannot_reclaim_or_change_capabilities(tmp_path: Path) -> None: + clock = [NOW] + queue = ObservatoryRecordedJobQueue( + tmp_path, definitions=_definitions(), clock=lambda: clock[0], claim_lease_seconds=10, + ) + job, _ = queue.submit(_intent(), enqueue=True) + args = {"claimant_id": "recorded-worker", "supported_executor_identities": _capabilities(), + "protocol_version": 3} + first = queue.claim_next(**args, claim_request_id="original-v3") + assert first is not None + with pytest.raises(ObservatoryRecordedQueueConflictError): + queue.claim_next(claimant_id="recorded-worker", claim_request_id="original-v3", + supported_executor_identities=(), protocol_version=3) + clock[0] = "2026-08-30T21:00:11.000Z" + with pytest.raises(ObservatoryRecordedQueueStaleClaimError): + queue.claim_next(**args, claim_request_id="original-v3") + second = queue.claim_next(**args, claim_request_id="new-v3") + assert second is not None and second.job.job_id == job.job_id + assert second.job.claim_generation == first.job.claim_generation + 1 + assert second.claim_token != first.claim_token + with pytest.raises(ObservatoryRecordedQueueStaleClaimError): + queue.start(job.job_id, claim_token=first.claim_token) + + +@pytest.mark.parametrize("version", [True, False, 1, 4, "3"]) +def test_claim_protocol_rejects_ambiguous_versions(tmp_path: Path, version) -> None: + with pytest.raises(ValueError, match="protocol version"): + _queue(tmp_path).claim_next(claimant_id="recorded-worker", claim_request_id="invalid", + supported_executor_identities=(), protocol_version=version) + + def test_submission_is_durable_exactly_idempotent_and_path_free(tmp_path: Path) -> None: queue = _queue(tmp_path) first, first_created = queue.submit(_intent()) diff --git a/tests/test_observatory_worker_api.py b/tests/test_observatory_worker_api.py index 4ee1c1b..03beb6a 100644 --- a/tests/test_observatory_worker_api.py +++ b/tests/test_observatory_worker_api.py @@ -1,6 +1,7 @@ from __future__ import annotations import hashlib +import sqlite3 from pathlib import Path from typing import Any, cast @@ -170,6 +171,35 @@ def _services_with_transient_publisher( return TestClient(app), queue, publisher +def test_v3_idle_is_explicit_and_grant_remains_idempotent(tmp_path: Path) -> None: + client, queue = _services(tmp_path) + readiness = client.get( + "/api/v1/worker/observatory/recorded-jobs/claim-readiness", headers=WORKER_HEADERS, + ) + assert readiness.status_code == 200 + assert readiness.json()["protocols"][1]["idle_policy"] == "non-binding-no-receipt" + assert client.get("/api/v1/worker/observatory/recorded-jobs/claim-readiness").status_code == 401 + request = {**_claim_request("v3-client"), "schema_version": CLAIM_SCHEMA.replace("/v2", "/v3")} + path = "/api/v1/worker/observatory/recorded-jobs/claims" + idle = client.post(path, json=request, headers=WORKER_HEADERS) + assert idle.status_code == 204 + assert idle.headers["X-Mission-Core-Idle-Claim"] == "non-binding" + with sqlite3.connect(queue.database_path) as connection: + assert connection.execute( + "SELECT COUNT(*) FROM observatory_recorded_claim_receipts" + ).fetchone()[0] == 0 + assert connection.execute( + "SELECT COUNT(*) FROM observatory_recorded_claim_grants_v3" + ).fetchone()[0] == 0 + job_id = _enqueue(queue) + grant = client.post(path, json=request, headers=WORKER_HEADERS) + assert grant.status_code == 200 + assert grant.json()["job"]["job_id"] == job_id + retry = client.post(path, json=request, headers=WORKER_HEADERS) + assert retry.status_code == 200 and retry.json() == grant.json() + assert client.post(path, json=request).status_code == 401 + + def _enqueue( queue: ObservatoryRecordedJobQueue, *, diff --git a/tests/test_observatory_worker_http_transport.py b/tests/test_observatory_worker_http_transport.py index 793d89e..d8aaa4a 100644 --- a/tests/test_observatory_worker_http_transport.py +++ b/tests/test_observatory_worker_http_transport.py @@ -95,6 +95,24 @@ def _claim_response() -> httpx.Response: ) +def test_gateway_uses_v3_without_falling_back_to_idle_receipt_growth(tmp_path: Path) -> None: + def handle(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + assert body["schema_version"] == "missioncore.observatory-worker-claim-request/v3" + assert body["supported_executor_identities"] == [] + return httpx.Response(204, headers={"X-Mission-Core-Idle-Claim": "non-binding"}) + + with ObservatoryWorkerHttpGateway( + base_url="http://127.0.0.1:8000", bearer_token=BEARER_TOKEN, + work_root=tmp_path, transport=httpx.MockTransport(handle), + ) as gateway: + for number in range(2): + assert gateway.claim_next( + claimant_id=WORKER_006_CONTOUR_ID, + claim_request_id=f"v3-idle-{number}", supported_executor_identities=(), + ) is None + + def _cache_claim(gateway: ObservatoryWorkerHttpGateway) -> None: payload = gateway.claim_next( claimant_id=WORKER_006_CONTOUR_ID,