From f05fc4671dbc37d6c114c660110e39d312c9ea54 Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Thu, 20 Aug 2026 00:22:15 -0400 Subject: [PATCH 1/3] feat(worker): claim-exempt secondmate message lane ops and the invariant carve message-put/message-collect as their own provider verbs dispatched like inventory: bounded, content-addressed, idempotent data-plane blob transfers in the slot's own state container, enforced to the session/ namespace inside the op functions. Controller commands verify the exact assigned worker read-only and never touch claims, leases, or controller.json. The carve sentence lands next to the claim contract in docs/azure-workers.md. --- bin/fm-azure-worker-provider.py | 225 ++++++++++++++++++++++++++++++++ bin/fm-worker-lifecycle.py | 107 +++++++++++++++ bin/fm-worker-lifecycle.sh | 4 +- docs/azure-workers.md | 1 + 4 files changed, 336 insertions(+), 1 deletion(-) diff --git a/bin/fm-azure-worker-provider.py b/bin/fm-azure-worker-provider.py index 18140c8b667..e87801badc7 100755 --- a/bin/fm-azure-worker-provider.py +++ b/bin/fm-azure-worker-provider.py @@ -53,6 +53,21 @@ # is a standalone pinned file that cannot import from the repository, so the # two literals are kept in step by a test rather than by runtime coupling. MAX_OUTCOME_BYTES = 256 * 1024 * 1024 +# The compartment message lane (message-put/message-collect) is the ONE +# provider operation family outside the per-slot claim contract: bounded, +# content-addressed, idempotent data-plane blob transfers that touch no +# compute, no money, and no lifecycle state. docs/azure-workers.md names the +# carve next to the claim contract; require_session_blob_name enforces its +# namespace boundary where it is used. MESSAGE_ATTACH_MAX_BYTES must equal the +# controller's constant of the same name (kept in step by a test). +MESSAGE_JSON_MAX_BYTES = 256 * 1024 +MESSAGE_ATTACH_MAX_BYTES = 256 * 1024 * 1024 +SESSION_BLOB_PREFIX = "session/" +MESSAGE_INBOX_PREFIX = "session/in/" +MESSAGE_ATTACH_PREFIX = "session/in/attach/" +MESSAGE_OUTBOX_PREFIX = "session/out/" +MESSAGE_COLLECT_MAX_BLOBS = 4096 +MESSAGE_LOCAL_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") # Staging (archive fetches plus the repository clone) runs BEFORE the wall # starts and collection runs after it ends, so a bound covering a whole guest # run is wall plus this, never wall alone. The GUEST gives up first and the @@ -1507,6 +1522,209 @@ def download_outcome_bundle(controller, account, container, name, expected_diges return len(body) +def require_session_blob_name(name, lane_prefix): + """Refuse any blob name outside the compartment session/ namespace. + + The message lane is the one claim-exempt provider operation family, and + this guard is where its boundary is ENFORCED rather than merely + documented: every blob name a message op touches must live under + session/ in the slot's own state container. message-put writes only + session/in/... and message-collect reads only session/out/..., so the + staging pair, the reservation record, and the outcome bundles are + structurally unreachable from the message ops; a name outside the + namespace, or a path-traversing alias for one, raises instead of being + touched. + """ + if ( + not isinstance(name, str) + or not name.startswith(SESSION_BLOB_PREFIX) + or ".." in name + or "\\" in name + or "\x00" in name + ): + raise ProviderError("message blob name is outside the session/ namespace") + if not name.startswith(lane_prefix): + raise ProviderError("message blob name is outside its {} lane".format(lane_prefix)) + return name + + +def verify_message_spec(message, required_field): + """Exact-shape check for one message-lane request from the controller.""" + if not isinstance(message, dict): + raise ProviderError("message lane request is malformed") + slot = message.get("slot") + if not isinstance(slot, int) or isinstance(slot, bool) or slot not in SKU_PLAN: + raise ProviderError("message lane slot is outside the reviewed sixteen") + bindings = message.get("bindings") + required_bindings = ( + "home_binding", "task", "task_generation", "assignment_generation", + "account_binding", "worktree_binding", "repository_binding", + "repository_generation", + ) + if not isinstance(bindings, dict) or any( + not isinstance(bindings.get(field), str) or not bindings[field] + for field in required_bindings + ): + raise ProviderError("message lane worker bindings are incomplete") + if not isinstance(message.get("cloud_generation"), int) or isinstance(message.get("cloud_generation"), bool): + raise ProviderError("message lane cloud generation is not exact") + if message.get("role") not in ("author", "secondmate"): + raise ProviderError("message lane worker role is not exact") + if not isinstance(message.get(required_field), str) or not message[required_field]: + raise ProviderError("message lane {} is absent".format(required_field)) + return slot + + +def message_put(controller, message): + """Upload one bounded, content-addressed message blob to the compartment + session inbox: session/in/.json for the JSON lane, or + session/in/attach/.bundle for the attachment lane. + + CLAIM-EXEMPT BY DESIGN: this op is dispatched like inventory, outside the + per-slot claim/lease/fence contract, because a leg's execute claim + occupies pending_actions[slot] for its whole wall and a claimed message + lane could never deliver during a leg. The exemption is safe only because + this op touches no compute, no money, and no lifecycle state: it never + runs a Run Command, never powers or deletes a resource, never edits + controller state, and writes exactly one blob whose name it derives from + the content digest. Idempotency comes from that content address: a replay + of the same content converges on the existing blob without a second + upload, and the same name holding different bytes refuses. Namespace + boundary: every name this op writes must sit under session/in/ - + require_session_blob_name raises on anything else. + """ + slot = verify_message_spec(message, "file") + lane = message.get("lane") + if lane not in ("json", "attach"): + raise ProviderError("message lane must be json or attach") + source = Path(message["file"]) + if source.is_symlink() or not source.is_file(): + raise ProviderError("message payload file is unavailable") + payload = source.read_bytes() + if not payload: + raise ProviderError("message payload is empty") + if lane == "json": + if len(payload) > MESSAGE_JSON_MAX_BYTES: + raise ProviderError( + "message payload exceeds its {}-byte bound".format(MESSAGE_JSON_MAX_BYTES)) + try: + json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + raise ProviderError("message payload is not valid JSON") + digest = hashlib.sha256(payload).hexdigest() + name = "{}{}.json".format(MESSAGE_INBOX_PREFIX, digest) + else: + if len(payload) > MESSAGE_ATTACH_MAX_BYTES: + raise ProviderError( + "message attachment exceeds its {}-byte bound".format(MESSAGE_ATTACH_MAX_BYTES)) + digest = hashlib.sha256(payload).hexdigest() + name = "{}{}.bundle".format(MESSAGE_ATTACH_PREFIX, digest) + require_session_blob_name(name, MESSAGE_INBOX_PREFIX) + storage = os.environ.get("FM_AZURE_STORAGE_NAME", "") + container = expected_names(controller, slot)["state-container"] + existing, rc, _stderr = az(controller, [ + "storage", "blob", "show", "--auth-mode", "login", "--account-name", storage, + "--container-name", container, "--name", name, + ], check=False) + if rc == 0 and isinstance(existing, dict): + properties = existing.get("properties", existing) or {} + length = properties.get("contentLength") + if length is None: + length = existing.get("contentLength") + if length != len(payload): + # The name IS the content digest, so a different size under it is + # corruption or a foreign writer, never a replay. + raise ProviderError( + "existing message blob {} differs from its content address".format(name)) + return {"blob_name": name, "sha256": digest, "bytes": len(payload), "replayed": True} + tags = action_tags(controller, { + "role": message["role"], "slot": slot, + "cloud_generation": message["cloud_generation"], "bindings": message["bindings"], + }) + uploaded = upload_bytes_blob(controller, storage, container, name, payload, tags) + if uploaded != digest: + raise ProviderError("message upload digest is not exact") + return {"blob_name": name, "sha256": digest, "bytes": len(payload), "replayed": False} + + +def message_collect(controller, message): + """Fetch new compartment outbox blobs (session/out/...) into one local + directory and report their names, sizes, and SHA-256 digests. + + CLAIM-EXEMPT BY DESIGN, read-only, and shaped like inventory: dumb + transport that touches no compute, no money, and no lifecycle state. It + performs NO chain verification - the secondmate monitor owns the + sequence/chain checks - and it never deletes or overwrites an existing + local file: an existing name with identical bytes is skipped, an existing + name with different bytes refuses the whole collect. Namespace boundary: + the LIST is prefixed to session/out/ and every returned name is + re-checked through require_session_blob_name, so a listing that names any + blob outside session/out/ refuses rather than fetching it. + """ + slot = verify_message_spec(message, "output_dir") + output_dir = Path(message["output_dir"]) + if output_dir.is_symlink() or not output_dir.is_dir(): + raise ProviderError("message collect output directory is unavailable") + storage = os.environ.get("FM_AZURE_STORAGE_NAME", "") + container = expected_names(controller, slot)["state-container"] + listing, rc, stderr = az(controller, [ + "storage", "blob", "list", "--auth-mode", "login", "--account-name", storage, + "--container-name", container, "--prefix", MESSAGE_OUTBOX_PREFIX, + "--num-results", str(MESSAGE_COLLECT_MAX_BLOBS), + ], check=False) + if rc != 0 or not isinstance(listing, list): + raise ProviderError("message outbox listing failed or was malformed: {}".format(stderr)) + if len(listing) >= MESSAGE_COLLECT_MAX_BLOBS: + raise ProviderError("message outbox exceeds its bounded listing") + fetched = [] + skipped = [] + for blob in sorted(listing, key=lambda item: str((item or {}).get("name", ""))): + if not isinstance(blob, dict): + raise ProviderError("message outbox listing entry is malformed") + name = require_session_blob_name(blob.get("name"), MESSAGE_OUTBOX_PREFIX) + local_name = name[len(MESSAGE_OUTBOX_PREFIX):] + if not MESSAGE_LOCAL_NAME.match(local_name): + raise ProviderError( + "message outbox blob name is unsupported: {}".format(str(name)[:200])) + properties = blob.get("properties", blob) or {} + length = properties.get("contentLength") + if length is None: + length = blob.get("contentLength") + if not isinstance(length, int) or isinstance(length, bool) or not 0 <= length <= MESSAGE_ATTACH_MAX_BYTES: + raise ProviderError("message outbox blob size is malformed or unbounded") + target = output_dir / local_name + if target.is_symlink(): + raise ProviderError( + "message collect refuses a symlinked local target: {}".format(local_name)) + fd, staging = tempfile.mkstemp(prefix="fm-message-collect-", dir=str(output_dir)) + os.close(fd) + try: + os.chmod(staging, 0o600) + _, download_rc, download_stderr = az(controller, [ + "storage", "blob", "download", "--auth-mode", "login", "--account-name", storage, + "--container-name", container, "--name", name, "--file", staging, "--overwrite", + ], check=False, timeout=AZ_TIMEOUT_SECONDS + length // (256 * 1024)) + if download_rc != 0: + raise ProviderError("message blob download failed: {}".format(download_stderr)) + body = Path(staging).read_bytes() + if len(body) != length: + raise ProviderError("message blob size differs from its listing claim") + digest = hashlib.sha256(body).hexdigest() + record = {"blob_name": name, "bytes": length, "sha256": digest} + if target.exists(): + if hashlib.sha256(target.read_bytes()).hexdigest() == digest: + skipped.append(record) + continue + raise ProviderError( + "collected message blob {} diverges from the existing local file".format(local_name)) + os.replace(staging, str(target)) + finally: + with contextlib.suppress(FileNotFoundError): + Path(staging).unlink() + fetched.append(record) + return {"fetched": fetched, "skipped": skipped} + + def staged_directory_archive(directory, manifest, label): """Deterministic tar of one flat staging directory, verified against the digest-bound request manifest before any byte leaves the controller.""" @@ -2508,6 +2726,13 @@ def main(): elif operation == "mutate": require_landed_code() value = response(controller, operation, result=mutate(controller, request.get("action"))) + elif operation in ("message-put", "message-collect"): + # The claim-exempt compartment message lane, dispatched like + # inventory: no landed-code gate (that gate owns compute mutations), + # no claim, no lease. The ops themselves bound payload size, require + # content-addressed names, and refuse any blob outside session/. + handler = message_put if operation == "message-put" else message_collect + value = response(controller, operation, result=handler(controller, request.get("action"))) else: raise ProviderError("provider operation is not supported") sys.stdout.buffer.write(canonical_bytes(value) + b"\n") diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index 6a3047a850e..e7f801151fd 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -96,8 +96,16 @@ # that actually ran. PROVIDER_GUEST_RUN_SLACK_SECONDS = 8400 MAX_PROVIDER_OUTPUT_BYTES = 2 * 1024 * 1024 +# The compartment message lane's attachment ceiling. Must equal +# MESSAGE_ATTACH_MAX_BYTES in bin/fm-azure-worker-provider.py, which owns the +# actual size refusal; this copy only sizes the provider subprocess deadline. +# Kept in step by a test rather than by runtime coupling. +MESSAGE_ATTACH_MAX_BYTES = 256 * 1024 * 1024 # Every provider mutation type the claim contract covers. admission-refused is # a bare planning verdict with no idempotency key, never claimed, never sent. +# message-put/message-collect are deliberately NOT here: the message lane is +# the one claim-exempt provider operation family (docs/azure-workers.md), and +# a message spec must never be storable as a pending claim. ACTION_TYPES = frozenset({ "create", "resume", "deallocate", "delete-compute", "reset", "execute", "steer", }) @@ -658,6 +666,12 @@ def provider_action_timeout(action): return PROVIDER_CREATE_TIMEOUT_SECONDS if action.get("type") == "steer": return PROVIDER_STEER_TIMEOUT_SECONDS + message_bytes = action.get("message_bytes") + if isinstance(message_bytes, int) and not isinstance(message_bytes, bool) and message_bytes >= 0: + # Message-lane blob transfers get a bound proportional to their + # declared size, mirroring the provider's own per-transfer az bounds; + # a 256 MiB attachment cannot move inside the ordinary bound. + return PROVIDER_TIMEOUT_SECONDS + message_bytes // (256 * 1024) return PROVIDER_TIMEOUT_SECONDS @@ -2121,6 +2135,25 @@ def parser(): steer.add_argument("--confirm-steer", action="store_true") steer.add_argument("--confirm-subscription", required=True) + message_put = sub.add_parser( + "message-put", + help="deliver one bounded content-addressed message blob to the compartment session inbox", + ) + message_put.add_argument("--task", required=True) + message_put.add_argument("--task-generation", required=True) + message_put.add_argument("--assignment-generation", required=True) + message_put.add_argument("--file", default=None, help="bounded JSON message payload") + message_put.add_argument("--attach", default=None, help="bounded binary attachment (child delta bundle)") + + message_collect = sub.add_parser( + "message-collect", + help="fetch new compartment session outbox blobs without verification or overwrite", + ) + message_collect.add_argument("--task", required=True) + message_collect.add_argument("--task-generation", required=True) + message_collect.add_argument("--assignment-generation", required=True) + message_collect.add_argument("--output-dir", required=True) + status = sub.add_parser("status", help="show bounded local lifecycle and cost evidence") status.add_argument("--live", action="store_true") status.add_argument("--json", action="store_true") @@ -3258,6 +3291,76 @@ def command_steer(env, args): print("steer request digest delivered to the exact worker generation") +def message_lane_worker(env, args, command): + """Resolve the exact assigned worker for one message-lane command. + + Read-only on purpose: the state is loaded under the fleet lock, checked + with command_execute's own identity gates (assigned status, exact + assignment generation, no release proof), and never saved - neither + message op modifies controller.json or any other lifecycle state. + """ + with controller_lock(env): + state = load_state(env) + key = request_key(require_id("task", args.task), require_id("task generation", args.task_generation)) + item = state["queue"].get(key) + if item is None or item.get("status") != "assigned": + raise LifecycleError("{} requires one exact assigned task generation".format(command)) + worker = state["workers"].get(str(item.get("slot"))) + if worker is None or worker.get("assignment_generation") != args.assignment_generation: + raise LifecycleError("{} assignment generation is not exact".format(command)) + if worker.get("release_proof") is not None: + raise LifecycleError("released work cannot use the compartment message lane") + return { + "slot": worker["slot"], + "role": worker.get("role", "author"), + "cloud_generation": worker["cloud_generation"], + "bindings": worker["bindings"], + } + + +def command_message_put(env, args): + if (args.file is None) == (args.attach is None): + raise LifecycleError("message-put requires exactly one of --file or --attach") + source = Path(args.file if args.file is not None else args.attach) + if source.is_symlink() or not source.is_file(): + raise LifecycleError("message payload file is unavailable: {}".format(source)) + message = message_lane_worker(env, args, "message-put") + message.update({ + "lane": "json" if args.file is not None else "attach", + "file": str(source.resolve()), + "message_bytes": source.stat().st_size, + }) + # THE ONE DELIBERATE CLAIM-EXEMPT CARVE (design R2/R3 B.1/B.9, and the + # doc sentence next to the claim contract in docs/azure-workers.md): no + # make_action, no claim_pending, no apply_pending, no slot_lease. A leg's + # execute claim occupies pending_actions[slot] for its whole wall and + # claim_pending refuses any different key on that slot, so a claimed + # message lane could never deliver during a leg - precisely when delivery + # matters. Safe only because the provider op is a bounded, + # content-addressed, idempotent data-plane blob write that touches no + # compute, no money, and no lifecycle state; idempotency comes from the + # content address, which is stronger than a claim for this payload class. + result = provider_call(env, "message-put", message)["result"] + print(json.dumps(result, sort_keys=True, separators=(",", ":"))) + + +def command_message_collect(env, args): + output_dir = Path(args.output_dir) + if output_dir.is_symlink() or not output_dir.is_dir(): + raise LifecycleError("message collect output directory is unavailable: {}".format(args.output_dir)) + message = message_lane_worker(env, args, "message-collect") + message.update({ + "output_dir": str(output_dir.resolve()), + "message_bytes": MESSAGE_ATTACH_MAX_BYTES, + }) + # Same claim-exempt carve as message-put: read-only dumb transport, + # shaped like inventory. Chain verification belongs to the secondmate + # monitor, never here, and the provider op refuses divergent overwrites + # of existing local files rather than deciding anything. + result = provider_call(env, "message-collect", message)["result"] + print(json.dumps(result, sort_keys=True, separators=(",", ":"))) + + def command_status(env, args): inventory = None if args.live: @@ -3328,6 +3431,10 @@ def main(argv=None): command_resume(env, args) elif args.command == "steer": command_steer(env, args) + elif args.command == "message-put": + command_message_put(env, args) + elif args.command == "message-collect": + command_message_collect(env, args) elif args.command == "status": command_status(env, args) else: diff --git a/bin/fm-worker-lifecycle.sh b/bin/fm-worker-lifecycle.sh index 883c4b78292..2db968bc78a 100755 --- a/bin/fm-worker-lifecycle.sh +++ b/bin/fm-worker-lifecycle.sh @@ -39,6 +39,8 @@ # fm-worker-lifecycle.sh surrender --task --task-generation --reason --output --confirm-surrender --confirm-subscription # fm-worker-lifecycle.sh resume # fm-worker-lifecycle.sh steer +# fm-worker-lifecycle.sh message-put --file | --attach +# fm-worker-lifecycle.sh message-collect --output-dir # fm-worker-lifecycle.sh status [--live] [--json] # fm-worker-lifecycle.sh acceptance-plan set -euo pipefail @@ -50,7 +52,7 @@ SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P) . "$SCRIPT_DIR/fm-cloud-state-lib.sh" case "${1:-}" in - request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release|abandon-claim) + request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release|abandon-claim|message-put|message-collect) fm_refuse_if_gate_agent exec python3 "$SCRIPT_DIR/fm-worker-lifecycle.py" "$@" ;; diff --git a/docs/azure-workers.md b/docs/azure-workers.md index 12858407cb8..4cbde306eaf 100644 --- a/docs/azure-workers.md +++ b/docs/azure-workers.md @@ -143,6 +143,7 @@ Provider-account credentials never enter ARM parameters, controller output, logs A submitted provider action has a canonical SHA-256 idempotency key and remains in the per-slot `pending_actions` map until the exact provider result is durably applied; the map entry is a deep copy that re-derives its own key at every load, and the legacy scalar slot permanently holds a sentinel an old binary refuses rather than misreads. After a host restart, the same action and key are replayed. +The compartment message lane (`message-put`/`message-collect`) is the one provider operation family outside the per-slot claim contract: bounded, content-addressed, idempotent data-plane blob transfers that touch no compute, no money, and no lifecycle state; every compute-mutating action keeps the full claim/lease/fence discipline. The Azure singleton deployment is incremental and receives the same task, home, assignment, and snapshot bindings, so replay converges one generation rather than creating a second assignment. A visible VM with another task or assignment binding refuses instead of being adopted. From 8af6823dc95e1fa8e425a7f5a720280a96300e36 Mon Sep 17 00:00:00 2001 From: Dongkeun Lee Date: Thu, 20 Aug 2026 00:27:54 -0400 Subject: [PATCH 2/3] test(worker): message lane bounds, claim exemption, and D.3 inventory pin Fixture provider grows the two message verbs; the real provider ops run hermetically against a stubbed az for bounds, content addressing, replay, namespace, and collect semantics; the e2e unit proves message-put succeeds across an outstanding execute claim while a fresh compute action refuses; the static unit pins inventory to exactly three named blob reads and the message commands to zero claim-machinery call sites. --- tests/fm-worker-lifecycle.test.sh | 580 +++++++++++++++++++++++++++++- 1 file changed, 579 insertions(+), 1 deletion(-) diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index d404a5e336d..e3eb9b2a26d 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -992,6 +992,49 @@ if request["operation"] == "mutate": raise AssertionError(kind) state["seen"][key] = result save() +elif request["operation"] in ("message-put", "message-collect"): + # The claim-exempt message lane: raw ops, never routed through mutate, + # never touching the mutate branch's idempotency-key machinery. Blobs + # live in fixture state as hex under session/... names per slot. + message = request["action"] + blobs = state.setdefault("session_blobs", {}).setdefault(str(message["slot"]), {}) + if request["operation"] == "message-put": + body = Path(message["file"]).read_bytes() + digest = hashlib.sha256(body).hexdigest() + if message["lane"] == "json": + name = "session/in/{}.json".format(digest) + else: + name = "session/in/attach/{}.bundle".format(digest) + replayed = name in blobs + if not replayed: + blobs[name] = body.hex() + state["calls"].append({"type": "message-put", "slot": message["slot"], "name": name}) + save() + result = {"blob_name": name, "sha256": digest, "bytes": len(body), "replayed": replayed} + else: + out = Path(message["output_dir"]) + fetched = [] + skipped = [] + for name in sorted(blobs): + if not name.startswith("session/out/"): + continue + body = bytes.fromhex(blobs[name]) + digest = hashlib.sha256(body).hexdigest() + target = out / name[len("session/out/"):] + record = {"blob_name": name, "bytes": len(body), "sha256": digest} + if target.exists(): + if hashlib.sha256(target.read_bytes()).hexdigest() == digest: + skipped.append(record) + continue + sys.stderr.write( + "FIXTURE PROVIDER REFUSED: collected message blob {} diverges " + "from the existing local file\n".format(target.name)) + raise SystemExit(1) + target.write_bytes(body) + fetched.append(record) + state["calls"].append({"type": "message-collect", "slot": message["slot"]}) + save() + result = {"fetched": fetched, "skipped": skipped} else: active = sum( 1 for worker in state["workers"].values() @@ -1031,7 +1074,7 @@ response = { "schema": "fm.worker-provider-response/v1", "operation": request["operation"], "controller": controller, } -response["result" if request["operation"] == "mutate" else "inventory"] = result +response["inventory" if request["operation"] == "inventory" else "result"] = result print(json.dumps(response, sort_keys=True, separators=(",", ":"))) PY chmod +x "$1" @@ -3654,6 +3697,538 @@ PY } +message_lane_provider_contract() { + local tmp + fm_test_tmproot_into tmp fm-message-lane-provider + python3 - "$AZURE" "$tmp" <<'PY' || fail "message lane provider contract failed" +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys + +provider_path, tmp = sys.argv[1], Path(sys.argv[2]) +os.environ["FM_AZURE_STORAGE_NAME"] = "stfmtestwkr01" +spec = importlib.util.spec_from_file_location("azure_provider", provider_path) +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + +SUB = "11111111-1111-4111-8111-111111111111" +controller = { + "home_binding": "0" * 64, "subscription": SUB, + "deployment_generation": "dep-one", "owner": "owner", "prefix": "fmtest", + "resource_group": "rg-fixture", +} +bindings = { + "home_binding": "1" * 64, "task": "task-msg", "task_generation": "gen-msg", + "assignment_generation": "asg-00000001", "account_binding": "2" * 64, + "worktree_binding": "3" * 64, "repository_binding": "4" * 64, + "repository_generation": "repo-msg", +} +def message(**fields): + base = {"slot": 3, "role": "author", "cloud_generation": 1, "bindings": bindings} + base.update(fields) + return base + +store = {} +uploads = [] + +def fake_az(ctrl, args, check=True, timeout=None): + if args[:3] == ["storage", "blob", "show"]: + name = args[args.index("--name") + 1] + if name not in store: + return None, 1, "BlobNotFound" + return {"properties": {"contentLength": len(store[name])}}, 0, "" + if args[:3] == ["storage", "blob", "upload"]: + name = args[args.index("--name") + 1] + store[name] = Path(args[args.index("--file") + 1]).read_bytes() + uploads.append(name) + return None, 0, "" + if args[:3] == ["storage", "blob", "list"]: + prefix = args[args.index("--prefix") + 1] + return ([ + {"name": name, "properties": {"contentLength": len(body)}} + for name, body in sorted(store.items()) if name.startswith(prefix) + ], 0, "") + if args[:3] == ["storage", "blob", "download"]: + Path(args[args.index("--file") + 1]).write_bytes(store[args[args.index("--name") + 1]]) + return None, 0, "" + raise AssertionError(args) + +module.az = fake_az + +def refuses(callable_, needle): + try: + callable_() + except module.ProviderError as exc: + assert needle in str(exc), (needle, str(exc)) + return + raise AssertionError("no refusal containing {!r}".format(needle)) + +# The exact reviewed bounds, and the exact refusal strings derived from them. +assert module.MESSAGE_JSON_MAX_BYTES == 262144 +assert module.MESSAGE_ATTACH_MAX_BYTES == 268435456 + +# JSON lane: content-addressed name, single upload, replay is a no-op success. +payload = json.dumps({"schema": "fm.secondmate-message/v1", "text": "hello"}).encode() +digest = hashlib.sha256(payload).hexdigest() +msg_file = tmp / "msg.json" +msg_file.write_bytes(payload) +result = module.message_put(controller, message(lane="json", file=str(msg_file))) +expected_name = "session/in/{}.json".format(digest) +assert result == {"blob_name": expected_name, "sha256": digest, "bytes": len(payload), "replayed": False}, result +assert uploads == [expected_name] and store[expected_name] == payload +replay = module.message_put(controller, message(lane="json", file=str(msg_file))) +assert replay["replayed"] is True and uploads == [expected_name], (replay, uploads) + +# Size bound, JSON requirement, and emptiness all refuse with exact strings +# BEFORE any az call could run. +big = tmp / "big.json" +big.write_bytes(json.dumps("a" * 262200).encode()) +refuses(lambda: module.message_put(controller, message(lane="json", file=str(big))), + "message payload exceeds its 262144-byte bound") +bad = tmp / "bad.json" +bad.write_bytes(b"not json {{{") +refuses(lambda: module.message_put(controller, message(lane="json", file=str(bad))), + "message payload is not valid JSON") +empty = tmp / "empty.json" +empty.write_bytes(b"") +refuses(lambda: module.message_put(controller, message(lane="json", file=str(empty))), + "message payload is empty") + +# Attach lane: binary content with NO JSON requirement, its own prefix, and a +# bound proven through the constant seam rather than a 256MiB fixture (the +# refusal string derives from the same constant the check reads). +blob = b"\x00\x01\x02binary-not-json\xff" * 8 +attach_file = tmp / "delta.bundle" +attach_file.write_bytes(blob) +attach_result = module.message_put(controller, message(lane="attach", file=str(attach_file))) +attach_digest = hashlib.sha256(blob).hexdigest() +assert attach_result["blob_name"] == "session/in/attach/{}.bundle".format(attach_digest), attach_result +original_bound = module.MESSAGE_ATTACH_MAX_BYTES +module.MESSAGE_ATTACH_MAX_BYTES = 4096 +try: + oversized = tmp / "oversized.bundle" + oversized.write_bytes(b"\xab" * 5000) + refuses(lambda: module.message_put(controller, message(lane="attach", file=str(oversized))), + "message attachment exceeds its 4096-byte bound") +finally: + module.MESSAGE_ATTACH_MAX_BYTES = original_bound + +# A name that already exists with bytes other than its own content address is +# corruption or a foreign writer, never a replay. +corrupt_payload = json.dumps({"n": 42}).encode() +corrupt_name = "session/in/{}.json".format(hashlib.sha256(corrupt_payload).hexdigest()) +store[corrupt_name] = b"xx" +corrupt_file = tmp / "corrupt.json" +corrupt_file.write_bytes(corrupt_payload) +refuses(lambda: module.message_put(controller, message(lane="json", file=str(corrupt_file))), + "differs from its content address") +del store[corrupt_name] + +# The namespace guard is the enforced boundary for BOTH ops. +assert module.require_session_blob_name("session/out/000001-aa.json", module.MESSAGE_OUTBOX_PREFIX) +refuses(lambda: module.require_session_blob_name("outcome-" + "a" * 32 + ".bundle", module.MESSAGE_OUTBOX_PREFIX), + "outside the session/ namespace") +refuses(lambda: module.require_session_blob_name("session/out/../../request.json", module.MESSAGE_OUTBOX_PREFIX), + "outside the session/ namespace") +refuses(lambda: module.require_session_blob_name("session/in/x.json", module.MESSAGE_OUTBOX_PREFIX), + "outside its session/out/ lane") + +# Collect: fetches new outbox blobs, never touches session/in/, skips +# identical existing files, refuses divergent ones, and reports exactly what +# it moved. +store.clear() +uploads.clear() +first = json.dumps({"sequence": 1}).encode() +second = json.dumps({"sequence": 2}).encode() +store["session/out/000001-aa.json"] = first +store["session/out/000002-bb.json"] = second +store["session/in/planted.json"] = b"{}" +outdir = tmp / "collected" +outdir.mkdir() +collected = module.message_collect(controller, message(output_dir=str(outdir))) +assert [entry["blob_name"] for entry in collected["fetched"]] == [ + "session/out/000001-aa.json", "session/out/000002-bb.json"], collected +assert collected["skipped"] == [] +assert sorted(path.name for path in outdir.iterdir()) == ["000001-aa.json", "000002-bb.json"] +assert (outdir / "000001-aa.json").read_bytes() == first +assert collected["fetched"][0]["sha256"] == hashlib.sha256(first).hexdigest() +again = module.message_collect(controller, message(output_dir=str(outdir))) +assert again["fetched"] == [] and len(again["skipped"]) == 2, again +(outdir / "000001-aa.json").write_bytes(b"locally diverged") +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "collected message blob 000001-aa.json diverges from the existing local file") +(outdir / "000001-aa.json").write_bytes(first) + +# A hostile or buggy listing cannot walk the op outside session/out/: foreign +# names, traversal aliases, nested paths, and unbounded sizes all refuse. +real_az = module.az +def hostile(entries): + def hostile_az(ctrl, args, check=True, timeout=None): + if args[:3] == ["storage", "blob", "list"]: + return entries, 0, "" + return real_az(ctrl, args, check=check, timeout=timeout) + return hostile_az +module.az = hostile([{"name": "outcome-evil.bundle", "properties": {"contentLength": 3}}]) +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "outside the session/ namespace") +module.az = hostile([{"name": "session/out/../request.json", "properties": {"contentLength": 3}}]) +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "outside the session/ namespace") +module.az = hostile([{"name": "session/out/nested/blob.json", "properties": {"contentLength": 3}}]) +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "message outbox blob name is unsupported") +module.az = hostile([{"name": "session/out/huge.bundle", "properties": {"contentLength": 268435457}}]) +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "message outbox blob size is malformed or unbounded") +module.az = real_az + +# A symlinked local target is never followed. +os.symlink(tmp / "elsewhere", outdir / "000003-cc.json") +store["session/out/000003-cc.json"] = b"{}" +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "symlinked local target") +(outdir / "000003-cc.json").unlink() +del store["session/out/000003-cc.json"] + +# End-to-end dispatch pin: the REAL provider binary routes message-put to the +# same bounded op, and the refusal fires before any az invocation exists to +# fail differently. +env = dict(os.environ) +env.update({ + "FM_AZURE_SUBSCRIPTION_ID": SUB, "FM_AZURE_DEPLOYMENT_GENERATION": "dep-one", + "FM_AZURE_OWNER_TAG": "owner", "FM_AZURE_NAMING_PREFIX": "fmtest", + "FM_AZURE_STORAGE_NAME": "stfmtestwkr01", +}) +request = { + "schema": "fm.worker-provider-request/v1", "operation": "message-put", + "controller": controller, "action": message(lane="json", file=str(big)), +} +proc = subprocess.run( + [sys.executable, provider_path], input=json.dumps(request).encode(), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, timeout=60, +) +assert proc.returncode == 2, (proc.returncode, proc.stderr) +assert b"AZURE WORKER PROVIDER REFUSED: message payload exceeds its 262144-byte bound" in proc.stderr, proc.stderr +PY + pass "message lane provider ops bound size, address by content, and hold the session/ boundary" +} + +message_lane_claim_exemption() { + local tmp provider fixture home envfile + fm_test_tmproot_into tmp fm-message-lane-claim + provider="$tmp/provider.py" + fixture="$tmp/provider-state.json" + home="$tmp/home" + mkdir -p "$home" + write_fixture_provider "$provider" + envfile="$tmp/env" + cat >"$envfile" < Date: Thu, 20 Aug 2026 01:20:52 -0400 Subject: [PATCH 3/3] fix(worker): collect never re-pays history and paginates by cursor; put replay proves the digest Adversarial review of the message lane: an existing local name is now judged without a transfer (digest metadata when stamped, exact size for digestless guest blobs), so a poll no longer re-downloads the whole outbox and cannot grow past the deadline; the 4096 hard refusal is replaced by a name-ordered --after cursor with a bounded marker walk, a per-call processing page, and a per-call transfer budget equal to the constant the subprocess deadline is sized from; put replay convergence now requires the stamped content_digest, refusing same-length different bytes and digestless foreign writers; the interim both-roles scope and the PR 4 assignment_generation delivery-fencing contract are stated in the docstrings. --- bin/fm-azure-worker-provider.py | 171 ++++++++++++++++++++++++----- bin/fm-worker-lifecycle.py | 15 +++ tests/fm-worker-lifecycle.test.sh | 176 +++++++++++++++++++++++++++--- 3 files changed, 319 insertions(+), 43 deletions(-) diff --git a/bin/fm-azure-worker-provider.py b/bin/fm-azure-worker-provider.py index e87801badc7..04fab2b745d 100755 --- a/bin/fm-azure-worker-provider.py +++ b/bin/fm-azure-worker-provider.py @@ -5,6 +5,13 @@ response. It never adopts names outside the reviewed resource group and never mutates a resource until complete owner, generation, task, cloud-instance, disk, account, and worktree identity matches the controller action. + +The compartment message lane (message-put/message-collect) is claim-exempt +data-plane transport with one delivery-fencing contract PR 4 must implement: +a slot-addressed transfer runs outside the controller lock, so a late message +can land in a recreated slot's container; the secondmate monitor therefore +stamps assignment_generation inside every message envelope and the session +runner refuses envelopes naming a foreign generation. """ import contextlib @@ -66,7 +73,19 @@ MESSAGE_INBOX_PREFIX = "session/in/" MESSAGE_ATTACH_PREFIX = "session/in/attach/" MESSAGE_OUTBOX_PREFIX = "session/out/" +# Collect is a bounded incremental walk, never a hard refusal on mailbox +# depth: at most MAX_BLOBS names per az listing page, at most MAX_PAGES pages +# walked per call while skipping already-collected history, at most +# PAGE_BLOBS entries processed per call, and at most the transfer budget +# downloaded per call. The budget equals the per-blob attach ceiling, so one +# maximum-size blob always fits in one call and the controller can size the +# subprocess deadline from the same number the fetch loop is bounded by. +# Anything beyond a bound is reported through the cursor and the more flag, +# collectable by the next call. MESSAGE_COLLECT_MAX_BLOBS = 4096 +MESSAGE_COLLECT_PAGE_BLOBS = 512 +MESSAGE_COLLECT_MAX_PAGES = 16 +MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES = MESSAGE_ATTACH_MAX_BYTES MESSAGE_LOCAL_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") # Staging (archive fetches plus the repository clone) runs BEFORE the wall # starts and collection runs after it ends, so a bound covering a whole guest @@ -1592,6 +1611,15 @@ def message_put(controller, message): upload, and the same name holding different bytes refuses. Namespace boundary: every name this op writes must sit under session/in/ - require_session_blob_name raises on anything else. + + Role scope: until PR 4's spawn lane creates compartments, no secondmate + worker exists to address, so the lane deliberately serves author-role + workers as well as secondmate compartments; PR 4/6 narrows the callers + to compartments. Delivery fencing (PR 4 contract): this transfer is + slot-addressed and runs outside the controller lock, so a late put can + land in a recreated slot's container; the monitor therefore stamps + assignment_generation inside every message envelope and the session + runner refuses envelopes naming a foreign generation. """ slot = verify_message_spec(message, "file") lane = message.get("lane") @@ -1631,9 +1659,13 @@ def message_put(controller, message): length = properties.get("contentLength") if length is None: length = existing.get("contentLength") - if length != len(payload): - # The name IS the content digest, so a different size under it is - # corruption or a foreign writer, never a replay. + metadata = existing.get("metadata") or properties.get("metadata") or {} + remote_digest = metadata.get("content_digest") or metadata.get("content-digest") + if remote_digest != digest or length != len(payload): + # The name IS the content digest and this op's own upload path + # always stamps content_digest metadata, so a missing or + # different digest under this name (same-length different bytes + # included) is corruption or a foreign writer, never a replay. raise ProviderError( "existing message blob {} differs from its content address".format(name)) return {"blob_name": name, "sha256": digest, "bytes": len(payload), "replayed": True} @@ -1647,40 +1679,100 @@ def message_put(controller, message): return {"blob_name": name, "sha256": digest, "bytes": len(payload), "replayed": False} +def message_outbox_listing(controller, storage, container, after): + """Name-ordered new outbox entries after the cursor, plus whether more + remain beyond this call's bounded walk. + + The az listing is name-ordered, so the cursor (a local outbox name) is a + deterministic high-water mark: entries at or before it are dropped + client-side, and when a deep already-collected history fills whole + listing pages the walk follows the service continuation marker for at + most MESSAGE_COLLECT_MAX_PAGES pages. The walk stops early once one full + processing page of new entries is in hand; anything beyond is reported + as more rather than refused. + """ + threshold = MESSAGE_OUTBOX_PREFIX + after if after else None + entries = [] + marker = None + for _page in range(MESSAGE_COLLECT_MAX_PAGES): + arguments = [ + "storage", "blob", "list", "--auth-mode", "login", "--account-name", storage, + "--container-name", container, "--prefix", MESSAGE_OUTBOX_PREFIX, + "--num-results", str(MESSAGE_COLLECT_MAX_BLOBS), "--include", "m", + "--show-next-marker", + ] + if marker: + arguments += ["--marker", marker] + listing, rc, stderr = az(controller, arguments, check=False) + if rc != 0 or not isinstance(listing, list): + raise ProviderError("message outbox listing failed or was malformed: {}".format(stderr)) + marker = None + page = [] + for item in listing: + if isinstance(item, dict) and "nextMarker" in item and "name" not in item: + marker = item.get("nextMarker") or None + continue + if not isinstance(item, dict): + raise ProviderError("message outbox listing entry is malformed") + page.append(item) + page.sort(key=lambda item: str(item.get("name", ""))) + for item in page: + if threshold is not None and str(item.get("name", "")) <= threshold: + continue + entries.append(item) + if len(entries) > MESSAGE_COLLECT_PAGE_BLOBS: + # One entry beyond the processing page proves more remain; + # the caller reports its cursor and the next call resumes. + return entries[:MESSAGE_COLLECT_PAGE_BLOBS], True + if marker is None: + return entries, False + return entries, True + + def message_collect(controller, message): """Fetch new compartment outbox blobs (session/out/...) into one local - directory and report their names, sizes, and SHA-256 digests. + directory and report their names, sizes, and SHA-256 digests, plus the + cursor (last processed local name) and whether more remain. CLAIM-EXEMPT BY DESIGN, read-only, and shaped like inventory: dumb transport that touches no compute, no money, and no lifecycle state. It performs NO chain verification - the secondmate monitor owns the sequence/chain checks - and it never deletes or overwrites an existing - local file: an existing name with identical bytes is skipped, an existing - name with different bytes refuses the whole collect. Namespace boundary: - the LIST is prefixed to session/out/ and every returned name is - re-checked through require_session_blob_name, so a listing that names any - blob outside session/out/ refuses rather than fetching it. + local file. It also never re-downloads collected history: an existing + local name is judged WITHOUT a transfer, against the listing's + content_digest metadata when the writer stamped it (this provider's own + uploads always do), or by exact size for digestless guest-written blobs; + a digest or size mismatch refuses the collect, and a digestless + same-size match is presumed already collected because the monitor's + chain verification owns full integrity. Each call downloads at most + MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES of new content and processes at + most one bounded page of entries after the optional cursor; the summary + reports the cursor and the more flag instead of ever hard-refusing a + deep mailbox. Namespace boundary: the LIST is prefixed to session/out/ + and every returned name is re-checked through require_session_blob_name, + so a listing that names any blob outside session/out/ refuses rather + than fetching it. + + Role scope: until PR 4's spawn lane creates compartments, no secondmate + worker exists to address, so the lane deliberately serves author-role + workers as well as secondmate compartments; PR 4/6 narrows the callers + to compartments. """ slot = verify_message_spec(message, "output_dir") output_dir = Path(message["output_dir"]) if output_dir.is_symlink() or not output_dir.is_dir(): raise ProviderError("message collect output directory is unavailable") + after = message.get("after") + if after is not None and (not isinstance(after, str) or not MESSAGE_LOCAL_NAME.match(after)): + raise ProviderError("message collect cursor is malformed") storage = os.environ.get("FM_AZURE_STORAGE_NAME", "") container = expected_names(controller, slot)["state-container"] - listing, rc, stderr = az(controller, [ - "storage", "blob", "list", "--auth-mode", "login", "--account-name", storage, - "--container-name", container, "--prefix", MESSAGE_OUTBOX_PREFIX, - "--num-results", str(MESSAGE_COLLECT_MAX_BLOBS), - ], check=False) - if rc != 0 or not isinstance(listing, list): - raise ProviderError("message outbox listing failed or was malformed: {}".format(stderr)) - if len(listing) >= MESSAGE_COLLECT_MAX_BLOBS: - raise ProviderError("message outbox exceeds its bounded listing") + entries, more = message_outbox_listing(controller, storage, container, after) fetched = [] skipped = [] - for blob in sorted(listing, key=lambda item: str((item or {}).get("name", ""))): - if not isinstance(blob, dict): - raise ProviderError("message outbox listing entry is malformed") + cursor = after + spent = 0 + for blob in entries: name = require_session_blob_name(blob.get("name"), MESSAGE_OUTBOX_PREFIX) local_name = name[len(MESSAGE_OUTBOX_PREFIX):] if not MESSAGE_LOCAL_NAME.match(local_name): @@ -1696,6 +1788,30 @@ def message_collect(controller, message): if target.is_symlink(): raise ProviderError( "message collect refuses a symlinked local target: {}".format(local_name)) + if target.exists(): + # Already-collected history is judged WITHOUT a transfer, or a + # poll would re-pay the whole outbox on every call and a deep + # history would eventually exceed any fixed deadline. + local_bytes = target.read_bytes() + local_digest = hashlib.sha256(local_bytes).hexdigest() + metadata = blob.get("metadata") or properties.get("metadata") or {} + remote_digest = metadata.get("content_digest") or metadata.get("content-digest") + if remote_digest is not None: + if remote_digest != local_digest: + raise ProviderError( + "collected message blob {} diverges from the existing local file".format(local_name)) + elif length != len(local_bytes): + raise ProviderError( + "collected message blob {} diverges from the existing local file".format(local_name)) + skipped.append({"blob_name": name, "bytes": length, "sha256": local_digest}) + cursor = local_name + continue + if spent and spent + length > MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES: + # Budget exhausted mid-walk: report the cursor and let the next + # call continue. The first fetch of a call always fits because + # no single blob exceeds the attach ceiling the budget equals. + more = True + break fd, staging = tempfile.mkstemp(prefix="fm-message-collect-", dir=str(output_dir)) os.close(fd) try: @@ -1710,19 +1826,14 @@ def message_collect(controller, message): if len(body) != length: raise ProviderError("message blob size differs from its listing claim") digest = hashlib.sha256(body).hexdigest() - record = {"blob_name": name, "bytes": length, "sha256": digest} - if target.exists(): - if hashlib.sha256(target.read_bytes()).hexdigest() == digest: - skipped.append(record) - continue - raise ProviderError( - "collected message blob {} diverges from the existing local file".format(local_name)) os.replace(staging, str(target)) finally: with contextlib.suppress(FileNotFoundError): Path(staging).unlink() - fetched.append(record) - return {"fetched": fetched, "skipped": skipped} + spent += length + fetched.append({"blob_name": name, "bytes": length, "sha256": digest}) + cursor = local_name + return {"fetched": fetched, "skipped": skipped, "cursor": cursor, "more": more} def staged_directory_archive(directory, manifest, label): diff --git a/bin/fm-worker-lifecycle.py b/bin/fm-worker-lifecycle.py index e7f801151fd..4983def35ca 100755 --- a/bin/fm-worker-lifecycle.py +++ b/bin/fm-worker-lifecycle.py @@ -2153,6 +2153,10 @@ def parser(): message_collect.add_argument("--task-generation", required=True) message_collect.add_argument("--assignment-generation", required=True) message_collect.add_argument("--output-dir", required=True) + message_collect.add_argument( + "--after", default=None, + help="resume after this local outbox name (the cursor a previous summary reported)", + ) status = sub.add_parser("status", help="show bounded local lifecycle and cost evidence") status.add_argument("--live", action="store_true") @@ -3298,6 +3302,11 @@ def message_lane_worker(env, args, command): with command_execute's own identity gates (assigned status, exact assignment generation, no release proof), and never saved - neither message op modifies controller.json or any other lifecycle state. + + Role scope: until PR 4's spawn lane creates compartments, no secondmate + worker exists to address, so the lane deliberately serves author-role + workers as well as secondmate compartments; PR 4/6 narrows the callers + to compartments. """ with controller_lock(env): state = load_state(env) @@ -3351,8 +3360,14 @@ def command_message_collect(env, args): message = message_lane_worker(env, args, "message-collect") message.update({ "output_dir": str(output_dir.resolve()), + # The provider caps each call's downloads at its transfer budget, + # which equals this constant, so the subprocess deadline is sized + # from the bytes one call can actually fetch; already-collected + # history is skipped without a transfer and costs nothing here. "message_bytes": MESSAGE_ATTACH_MAX_BYTES, }) + if args.after is not None: + message["after"] = args.after # Same claim-exempt carve as message-put: read-only dumb transport, # shaped like inventory. Chain verification belongs to the secondmate # monitor, never here, and the provider op refuses divergent overwrites diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh index e3eb9b2a26d..eef6039fed9 100755 --- a/tests/fm-worker-lifecycle.test.sh +++ b/tests/fm-worker-lifecycle.test.sh @@ -1013,18 +1013,24 @@ elif request["operation"] in ("message-put", "message-collect"): result = {"blob_name": name, "sha256": digest, "bytes": len(body), "replayed": replayed} else: out = Path(message["output_dir"]) + after = message.get("after") fetched = [] skipped = [] + cursor = after for name in sorted(blobs): if not name.startswith("session/out/"): continue + local_name = name[len("session/out/"):] + if after is not None and local_name <= after: + continue body = bytes.fromhex(blobs[name]) digest = hashlib.sha256(body).hexdigest() - target = out / name[len("session/out/"):] + target = out / local_name record = {"blob_name": name, "bytes": len(body), "sha256": digest} if target.exists(): if hashlib.sha256(target.read_bytes()).hexdigest() == digest: skipped.append(record) + cursor = local_name continue sys.stderr.write( "FIXTURE PROVIDER REFUSED: collected message blob {} diverges " @@ -1032,9 +1038,10 @@ elif request["operation"] in ("message-put", "message-collect"): raise SystemExit(1) target.write_bytes(body) fetched.append(record) + cursor = local_name state["calls"].append({"type": "message-collect", "slot": message["slot"]}) save() - result = {"fetched": fetched, "skipped": skipped} + result = {"fetched": fetched, "skipped": skipped, "cursor": cursor, "more": False} else: active = sum( 1 for worker in state["workers"].values() @@ -3734,13 +3741,21 @@ def message(**fields): store = {} uploads = [] +downloads = {"count": 0} + +def entry_for(name, body): + return { + "name": name, + "properties": {"contentLength": len(body)}, + "metadata": {"content_digest": hashlib.sha256(body).hexdigest()}, + } def fake_az(ctrl, args, check=True, timeout=None): if args[:3] == ["storage", "blob", "show"]: name = args[args.index("--name") + 1] if name not in store: return None, 1, "BlobNotFound" - return {"properties": {"contentLength": len(store[name])}}, 0, "" + return entry_for(name, store[name]), 0, "" if args[:3] == ["storage", "blob", "upload"]: name = args[args.index("--name") + 1] store[name] = Path(args[args.index("--file") + 1]).read_bytes() @@ -3748,11 +3763,15 @@ def fake_az(ctrl, args, check=True, timeout=None): return None, 0, "" if args[:3] == ["storage", "blob", "list"]: prefix = args[args.index("--prefix") + 1] - return ([ - {"name": name, "properties": {"contentLength": len(body)}} - for name, body in sorted(store.items()) if name.startswith(prefix) - ], 0, "") + limit = int(args[args.index("--num-results") + 1]) + names = [name for name in sorted(store) if name.startswith(prefix)] + start = int(args[args.index("--marker") + 1]) if "--marker" in args else 0 + page = [entry_for(name, store[name]) for name in names[start:start + limit]] + if "--show-next-marker" in args and start + limit < len(names): + page.append({"nextMarker": str(start + limit)}) + return page, 0, "" if args[:3] == ["storage", "blob", "download"]: + downloads["count"] += 1 Path(args[args.index("--file") + 1]).write_bytes(store[args[args.index("--name") + 1]]) return None, 0, "" raise AssertionError(args) @@ -3818,14 +3837,30 @@ finally: module.MESSAGE_ATTACH_MAX_BYTES = original_bound # A name that already exists with bytes other than its own content address is -# corruption or a foreign writer, never a replay. +# corruption or a foreign writer, never a replay. The judgment is the stamped +# content_digest metadata, so a SAME-LENGTH different-bytes blob refuses too, +# and a blob with no digest metadata at all (a foreign writer; this op's own +# uploads always stamp it) refuses rather than reading as a replay. corrupt_payload = json.dumps({"n": 42}).encode() corrupt_name = "session/in/{}.json".format(hashlib.sha256(corrupt_payload).hexdigest()) -store[corrupt_name] = b"xx" corrupt_file = tmp / "corrupt.json" corrupt_file.write_bytes(corrupt_payload) +store[corrupt_name] = b"xx" refuses(lambda: module.message_put(controller, message(lane="json", file=str(corrupt_file))), "differs from its content address") +store[corrupt_name] = b"Y" * len(corrupt_payload) +refuses(lambda: module.message_put(controller, message(lane="json", file=str(corrupt_file))), + "differs from its content address") +def digestless_show(ctrl, args, check=True, timeout=None): + if args[:3] == ["storage", "blob", "show"]: + name = args[args.index("--name") + 1] + return {"properties": {"contentLength": len(store[name])}}, 0, "" + return fake_az(ctrl, args, check=check, timeout=timeout) +store[corrupt_name] = corrupt_payload +module.az = digestless_show +refuses(lambda: module.message_put(controller, message(lane="json", file=str(corrupt_file))), + "differs from its content address") +module.az = fake_az del store[corrupt_name] # The namespace guard is the enforced boundary for BOTH ops. @@ -3837,11 +3872,13 @@ refuses(lambda: module.require_session_blob_name("session/out/../../request.json refuses(lambda: module.require_session_blob_name("session/in/x.json", module.MESSAGE_OUTBOX_PREFIX), "outside its session/out/ lane") -# Collect: fetches new outbox blobs, never touches session/in/, skips -# identical existing files, refuses divergent ones, and reports exactly what -# it moved. +# Collect: fetches new outbox blobs, never touches session/in/, and NEVER +# re-downloads collected history: an existing local name is judged against +# the listing's digest metadata without a transfer (the download counter is +# the proof), identical skips, divergent refuses. store.clear() uploads.clear() +downloads["count"] = 0 first = json.dumps({"sequence": 1}).encode() second = json.dumps({"sequence": 2}).encode() store["session/out/000001-aa.json"] = first @@ -3853,15 +3890,106 @@ collected = module.message_collect(controller, message(output_dir=str(outdir))) assert [entry["blob_name"] for entry in collected["fetched"]] == [ "session/out/000001-aa.json", "session/out/000002-bb.json"], collected assert collected["skipped"] == [] +assert collected["cursor"] == "000002-bb.json" and collected["more"] is False, collected +assert downloads["count"] == 2, downloads assert sorted(path.name for path in outdir.iterdir()) == ["000001-aa.json", "000002-bb.json"] assert (outdir / "000001-aa.json").read_bytes() == first assert collected["fetched"][0]["sha256"] == hashlib.sha256(first).hexdigest() again = module.message_collect(controller, message(output_dir=str(outdir))) assert again["fetched"] == [] and len(again["skipped"]) == 2, again +assert again["cursor"] == "000002-bb.json" and again["more"] is False, again +assert downloads["count"] == 2, ("collected history was re-downloaded", downloads) +# Divergence is decided from the digest metadata, also without a transfer. (outdir / "000001-aa.json").write_bytes(b"locally diverged") refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), "collected message blob 000001-aa.json diverges from the existing local file") +assert downloads["count"] == 2, ("a divergence check downloaded the blob", downloads) (outdir / "000001-aa.json").write_bytes(first) +# The cursor makes the walk incremental: nothing at or before it is touched. +after_cursor = module.message_collect(controller, message(output_dir=str(outdir), after="000001-aa.json")) +assert after_cursor["fetched"] == [] and len(after_cursor["skipped"]) == 1, after_cursor +assert after_cursor["skipped"][0]["blob_name"] == "session/out/000002-bb.json", after_cursor +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir), after="../evil")), + "message collect cursor is malformed") + +# Digestless guest-written blobs (no metadata) are judged by exact size: +# same size skips without a transfer, a size mismatch refuses. +def digestless_list(entries): + def digestless_az(ctrl, args, check=True, timeout=None): + if args[:3] == ["storage", "blob", "list"]: + return entries, 0, "" + return fake_az(ctrl, args, check=check, timeout=timeout) + return digestless_az +module.az = digestless_list([ + {"name": "session/out/000001-aa.json", "properties": {"contentLength": len(first)}}, +]) +digestless = module.message_collect(controller, message(output_dir=str(outdir))) +assert digestless["fetched"] == [] and len(digestless["skipped"]) == 1, digestless +assert downloads["count"] == 2, ("a digestless same-size blob was re-downloaded", downloads) +module.az = digestless_list([ + {"name": "session/out/000001-aa.json", "properties": {"contentLength": len(first) + 7}}, +]) +refuses(lambda: module.message_collect(controller, message(output_dir=str(outdir))), + "collected message blob 000001-aa.json diverges from the existing local file") +assert downloads["count"] == 2, downloads +module.az = fake_az + +# Cursor pagination collects a mailbox deeper than one call's processing +# page across successive calls (proven through the constant seam), and the +# per-call transfer budget stops a call early with an honest cursor. +paged_store_names = ["session/out/{:08d}-pp.json".format(index) for index in range(1, 6)] +store.clear() +for index, name in enumerate(paged_store_names, 1): + store[name] = json.dumps({"page_sequence": index}).encode() +paged_dir = tmp / "collected-paged" +paged_dir.mkdir() +original_page = module.MESSAGE_COLLECT_PAGE_BLOBS +module.MESSAGE_COLLECT_PAGE_BLOBS = 2 +try: + page_one = module.message_collect(controller, message(output_dir=str(paged_dir))) + assert len(page_one["fetched"]) == 2 and page_one["more"] is True, page_one + assert page_one["cursor"] == "00000002-pp.json", page_one + page_two = module.message_collect(controller, message(output_dir=str(paged_dir), after=page_one["cursor"])) + assert len(page_two["fetched"]) == 2 and page_two["more"] is True, page_two + page_three = module.message_collect(controller, message(output_dir=str(paged_dir), after=page_two["cursor"])) + assert len(page_three["fetched"]) == 1 and page_three["more"] is False, page_three +finally: + module.MESSAGE_COLLECT_PAGE_BLOBS = original_page +assert sorted(path.name for path in paged_dir.iterdir()) == [name.split("/")[-1] for name in paged_store_names] + +# The marker walk crosses truncated listing pages inside one call: with a +# two-entry listing page the whole five-blob mailbox still collects at once. +marker_dir = tmp / "collected-marker" +marker_dir.mkdir() +original_max = module.MESSAGE_COLLECT_MAX_BLOBS +module.MESSAGE_COLLECT_MAX_BLOBS = 2 +try: + marker_walk = module.message_collect(controller, message(output_dir=str(marker_dir))) + assert len(marker_walk["fetched"]) == 5 and marker_walk["more"] is False, marker_walk +finally: + module.MESSAGE_COLLECT_MAX_BLOBS = original_max + +# The transfer budget bounds one call's downloads and reports the remainder +# through the cursor instead of refusing or overrunning. +budget_dir = tmp / "collected-budget" +budget_dir.mkdir() +store.clear() +store["session/out/00000001-bg.json"] = b"12345678" +store["session/out/00000002-bg.json"] = b"87654321" +original_budget = module.MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES +module.MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES = 10 +try: + budget_one = module.message_collect(controller, message(output_dir=str(budget_dir))) + assert len(budget_one["fetched"]) == 1 and budget_one["more"] is True, budget_one + assert budget_one["cursor"] == "00000001-bg.json", budget_one + budget_two = module.message_collect(controller, message(output_dir=str(budget_dir), after=budget_one["cursor"])) + assert len(budget_two["fetched"]) == 1 and budget_two["more"] is False, budget_two +finally: + module.MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES = original_budget + +store.clear() +store["session/out/000001-aa.json"] = first +store["session/out/000002-bb.json"] = second # A hostile or buggy listing cannot walk the op outside session/out/: foreign # names, traversal aliases, nested paths, and unbounded sizes all refuse. @@ -4043,6 +4171,14 @@ again = json.loads(run( "--assignment-generation", worker1["assignment_generation"], "--output-dir", str(outdir), ).stdout) assert again["fetched"] == [] and len(again["skipped"]) == 2, again +assert again["cursor"] == "000002-bb.json" and again["more"] is False, again +resumed = json.loads(run( + "message-collect", "--task", "task-1", "--task-generation", "gen-1", + "--assignment-generation", worker1["assignment_generation"], "--output-dir", str(outdir), + "--after", "000001-aa.json", +).stdout) +assert resumed["fetched"] == [] and len(resumed["skipped"]) == 1, resumed +assert resumed["skipped"][0]["blob_name"] == "session/out/000002-bb.json", resumed (outdir / "000001-aa.json").write_bytes(b"locally diverged") diverged = run( "message-collect", "--task", "task-1", "--task-generation", "gen-1", @@ -4100,7 +4236,8 @@ collect_exempt = json.loads(run( "message-collect", "--task", "task-2", "--task-generation", "gen-2", "--assignment-generation", worker2["assignment_generation"], "--output-dir", str(outdir2), ).stdout) -assert collect_exempt == {"fetched": [], "skipped": []}, collect_exempt +assert collect_exempt["fetched"] == [] and collect_exempt["skipped"] == [], collect_exempt +assert "cursor" in collect_exempt and collect_exempt["more"] is False, collect_exempt message_calls = [entry for entry in fixture_state()["calls"] if entry["type"] == "message-put"] assert any(entry["slot"] == item2["slot"] for entry in message_calls), message_calls @@ -4181,11 +4318,24 @@ for op_name, lane_marker in (("message_put", "session/in/"), ("message_collect", docstring = ast.get_docstring(op_node) or "" assert "CLAIM-EXEMPT" in docstring, op_name assert lane_marker in docstring, op_name + # The interim role scope (both roles until PR 4 spawns compartments) is + # stated where the ops live, not discovered in production. + assert "author-role" in docstring and "PR 4/6" in docstring, op_name assert "require_session_blob_name" in segment(azure_src, op_name), op_name collect_doc = ast.get_docstring(node_of(azure_src, "message_collect")) or "" assert "never deletes or overwrites" in collect_doc +assert "never re-downloads collected history" in collect_doc +put_doc = ast.get_docstring(node_of(azure_src, "message_put")) or "" +assert "assignment_generation" in put_doc, "the delivery-fencing contract left the put docstring" +module_doc = ast.get_docstring(ast.parse(azure_src)) or "" +assert "assignment_generation" in module_doc, "the delivery-fencing contract left the module docstring" guard_doc = ast.get_docstring(node_of(azure_src, "require_session_blob_name")) or "" assert "session/" in guard_doc and "ENFORCED" in guard_doc +# The per-call transfer budget IS the constant the controller sizes the +# subprocess deadline from, and collect never hard-refuses mailbox depth. +assert "MESSAGE_COLLECT_TRANSFER_BUDGET_BYTES = MESSAGE_ATTACH_MAX_BYTES" in azure_src +collect_src = segment(azure_src, "message_collect") +assert "message outbox exceeds" not in collect_src, "the mailbox-depth hard refusal came back" # Controller side of the carve: the message commands never touch the claim # machinery or the durable document, and the generic mutate verb still