fix(observatory): version non-binding idle claims and preserve grant history

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()

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(
+            receipts = connection.execute(
-                "SELECT * FROM observatory_recorded_claim_receipts WHERE claim_request_id = ?",
+                "SELECT * FROM observatory_recorded_claim_receipts WHERE claim_request_id = ? "
-                (claim_request_id,),
+                "UNION ALL SELECT * FROM observatory_recorded_claim_grants_v3 "
-            ).fetchone()
+                "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(
+            receipt_table = (
-                connection,
+                "observatory_recorded_claim_grants_v3"
-                table="observatory_recorded_claim_receipts",
+                if protocol_version == 3
-                limit=self._max_claim_receipts,
+                else "observatory_recorded_claim_receipts"
-                label="claim receipt",
             )
+            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",

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()

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}")

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()

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())

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,
     *,

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,