fix(worker): admit the compartment payload pair so a secondmate leg can dispatch

bin/fm-worker-lifecycle.py

@@ -3558,13 +3558,49 @@ def command_capacity_release(env, args):
     print("specialized capacity reservation released after exact zero-compute proof")
 
 
+# What an ORDINARY crewmate payload may contain: the repository as a
+# credential-free bundle, plus the one task file its entrypoint reads. This set
+# is deliberately NOT widened for the compartment lane; see below.
 PAYLOAD_FILE_BOUNDS = {
     "repo.bundle": 512 * 1024 * 1024,
     "brief.md": 256 * 1024,
 }
+PAYLOAD_REQUIRED = ("repo.bundle", "brief.md")
+# What a SECONDMATE COMPARTMENT payload may contain: the ordinary set plus the
+# two files bin/fm-spawn.sh stages only for KIND=secondmate - the session runner
+# and the spawn-intent pi extension, which the compartment monitor's leg argv
+# names by path at /mnt/task/.fm-task/. Bounds are the smallest round numbers
+# leaving real headroom over the measured sizes (45142 B and 3867 B on the
+# azaccept compartment): a bound is a security control, so headroom buys against
+# ordinary source growth, not against a file becoming a different KIND of thing.
+COMPARTMENT_PAYLOAD_FILE_BOUNDS = {
+    **PAYLOAD_FILE_BOUNDS,
+    "fm-secondmate-session.py": 256 * 1024,
+    "fm-secondmate-spawn.pi-ext.ts": 64 * 1024,
+}
+# Both are REQUIRED, not merely admitted: the leg argv runs the runner and
+# passes --pi-ext, so a compartment whose staging silently lost either file
+# would dispatch a leg that cannot work. Refuse at the controller instead.
+COMPARTMENT_PAYLOAD_REQUIRED = PAYLOAD_REQUIRED + (
+    "fm-secondmate-session.py",
+    "fm-secondmate-spawn.pi-ext.ts",
+)
 ACCOUNT_TOTAL_BOUND = 1024 * 1024
 
 
+def payload_contract(role):
+    """The one owner of "what may a payload for this lane contain".
+
+    Returns (bounds, required) for the worker's durable role. Splitting by lane
+    rather than flattening one set keeps the ordinary crewmate lane exactly as
+    narrow as it is today: an author worker that somehow staged a session
+    runner is still refused.
+    """
+    if role == "secondmate":
+        return COMPARTMENT_PAYLOAD_FILE_BOUNDS, COMPARTMENT_PAYLOAD_REQUIRED
+    return PAYLOAD_FILE_BOUNDS, PAYLOAD_REQUIRED
+
+
 def staged_directory_manifest(label, directory, bounds=None, total_bound=None, required=()):
     """Digest one flat staging directory into {name: {sha256, bytes}}.
 
@@ -3626,14 +3662,6 @@ def command_execute(env, args):
         if outcome_root.is_symlink() or not outcome_root.is_dir():
             raise LifecycleError("outcome directory is unavailable: {}".format(args.outcome_dir))
     payload_manifest = account_manifest = None
-    if args.payload_dir is not None:
-        payload_manifest = staged_directory_manifest(
-            "payload", args.payload_dir, bounds=PAYLOAD_FILE_BOUNDS,
-            required=("repo.bundle", "brief.md"),
-        )
-        account_manifest = staged_directory_manifest(
-            "account", args.account_dir, total_bound=ACCOUNT_TOTAL_BOUND,
-        )
     inventory = provider_call(env, "inventory")["inventory"]
     with contextlib.ExitStack() as stack:
       with controller_lock(env):
@@ -3651,6 +3679,24 @@ def command_execute(env, args):
         classification, reason = classify_worker(worker, cloud)
         if classification != "assigned":
             raise LifecycleError("execute refuses a non-assigned or ambiguous worker: {}".format(reason))
+        if args.payload_dir is not None:
+            # The staging contract is chosen by the worker's DURABLE role, so it
+            # is resolved here (under the one controller lock, with the queue
+            # item and worker record in hand) rather than from the payload's own
+            # contents - a payload must never select the rules it is judged by.
+            # create_worker_record copies the item's role onto the worker, so a
+            # disagreement means the two records have drifted: fail closed.
+            worker_role = worker.get("role", "author")
+            if worker_role != item.get("role", "author"):
+                raise LifecycleError(
+                    "execute refuses a worker whose role disagrees with its queue item")
+            bounds, required = payload_contract(worker_role)
+            payload_manifest = staged_directory_manifest(
+                "payload", args.payload_dir, bounds=bounds, required=required,
+            )
+            account_manifest = staged_directory_manifest(
+                "account", args.account_dir, total_bound=ACCOUNT_TOTAL_BOUND,
+            )
         request = {
             "schema": EXECUTION_SCHEMA,
             **worker["bindings"],

tests/fm-secondmate-cloud-monitor.test.sh

@@ -2898,6 +2898,42 @@ PY
   assert_present "$SP_HOME/state/helm.cloud-payload/brief.md" "payload lacks the brief"
   assert_present "$SP_HOME/state/helm.cloud-payload/fm-secondmate-session.py" "payload lacks the session runner"
   assert_present "$SP_HOME/state/helm.cloud-payload/fm-secondmate-spawn.pi-ext.ts" "payload lacks the pi extension"
+  # EFFECT-shaped, not syntax-shaped: run the REAL lifecycle validator over the
+  # REAL directory the REAL fm-spawn.sh just produced. The assertions above name
+  # files they expect to be present; this one bounds what may be present at all,
+  # so a producer that stages an unadmitted file goes red here no matter HOW the
+  # staging was spelled (literal name, shell variable, trailing-slash cp). That
+  # distinction matters: the defect this closes was a producer and a validator
+  # drifting apart, and a guard that recognizes only today's syntax is the same
+  # class of weakness as the defect.
+  python3 - "$SP_HOME/state/helm.cloud-payload" "$ROOT/bin/fm-worker-lifecycle.py" \
+    <<'PY' || fail "the staged compartment payload is not admitted by the reviewed set"
+import importlib.util
+import sys
+from pathlib import Path
+
+payload = Path(sys.argv[1])
+spec = importlib.util.spec_from_file_location("lifecycle", sys.argv[2])
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+bounds, required = module.payload_contract("secondmate")
+
+# Deliberately NOT filtering dotfiles, though staged_directory_manifest skips
+# them: nothing should ever put a dotfile here, so this is stricter than the
+# validator on purpose and a stray one goes red instead of travelling unseen.
+staged = sorted(entry.name for entry in payload.iterdir())
+unadmitted = [name for name in staged if name not in bounds]
+assert not unadmitted, (
+    "fm-spawn.sh staged {} into the compartment payload, which the reviewed set "
+    "does not admit; every leg dispatch would refuse. Staged: {}".format(
+        unadmitted, staged))
+
+# The validator itself is the authority, so this cannot drift from what the
+# controller enforces at dispatch: it bounds bytes and requires the pair too.
+manifest = module.staged_directory_manifest(
+    "payload", payload, bounds=bounds, required=required)
+assert sorted(manifest) == staged, (sorted(manifest), staged)
+PY
   assert_present "$SP_HOME/state/helm.cloud-account/auth.json" "account staging lacks the auth projection"
   # Durable leg config rode into the persisted compartment environment.
   assert_grep 'FM_SECONDMATE_LEG_SECONDS=7200' "$SP_HOME/state/helm.cloud-env" "leg config was not persisted for the monitor"