feat(worker): claim-exempt secondmate message lane (R2/R3 PR 3)

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
@@ -53,6 +60,33 @@
 # 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/"
+# 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
 # run is wall plus this, never wall alone. The GUEST gives up first and the
@@ -1507,6 +1541,301 @@ 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/<sha256>.json for the JSON lane, or
+    session/in/attach/<sha256>.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.
+
+    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")
+    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")
+        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}
+    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_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, 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. 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"]
+    entries, more = message_outbox_listing(controller, storage, container, after)
+    fetched = []
+    skipped = []
+    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):
+            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))
+        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:
+            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()
+            os.replace(staging, str(target))
+        finally:
+            with contextlib.suppress(FileNotFoundError):
+                Path(staging).unlink()
+        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):
     """Deterministic tar of one flat staging directory, verified against the
     digest-bound request manifest before any byte leaves the controller."""
@@ -2508,6 +2837,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")