fix(azure): recover repeated worker executions

Author: ruby-dlee

bin/fm-azure-worker-provider.py

@@ -1595,6 +1595,32 @@ def upload_json_blob(
                         and hashlib.sha256(current_payload).hexdigest() == digest
                     ):
                         return digest
+                    # Sequential executes deliberately reuse the assignment's
+                    # fixed request/result blob names. The worker record can
+                    # still carry the ETag from the prior execute, so adopt the
+                    # observed same-assignment blob and retry one CAS rather
+                    # than permanently wedging retained-disk recovery. Another
+                    # assignment never passes the complete expected tag subset,
+                    # and a concurrent writer loses the fresh If-Match.
+                    current_metadata = current.get("metadata") or {}
+                    expected_metadata = tags_to_metadata(tags)
+                    if overwrite and all(
+                        current_metadata.get(key) == item
+                        for key, item in expected_metadata.items()
+                    ):
+                        retry_args = list(upload_args)
+                        match_index = retry_args.index("--if-match")
+                        retry_args[match_index + 1] = current_etag
+                        _, retry_rc, retry_stderr = az(
+                            controller, retry_args, check=False
+                        )
+                        if retry_rc == 0:
+                            return digest
+                        raise ProviderError(
+                            "conditionally retried worker staging upload failed: {}".format(
+                                retry_stderr
+                            )
+                        )
                 finally:
                     with contextlib.suppress(FileNotFoundError):
                         Path(current_path).unlink()
@@ -2546,6 +2572,24 @@ def mutate_deallocate(controller, action):
     return final
 
 
+def service_cancel_allows_missing_task_command(action):
+    proof = action.get("service_cancel_proof")
+    if not isinstance(proof, dict):
+        return False
+    unsigned = dict(proof)
+    supplied = unsigned.pop("proof_digest", None)
+    bindings = action.get("bindings") or {}
+    return (
+        proof.get("schema") == "fm.worker-service-cancel/v1"
+        and proof.get("verdict") == "cancelled-before-execution"
+        and supplied == hashlib.sha256(canonical_bytes(unsigned)).hexdigest()
+        and proof.get("task") == bindings.get("task")
+        and proof.get("task_generation") == bindings.get("task_generation")
+        and proof.get("assignment_generation") == bindings.get("assignment_generation")
+        and proof.get("cloud_instance_id") == action.get("cloud_instance_id")
+    )
+
+
 def mutate_delete_compute(controller, action):
     snapshot = inventory(controller, include_metrics=False)
     worker = worker_by_slot(snapshot, action["slot"])
@@ -2566,7 +2610,11 @@ def mutate_delete_compute(controller, action):
             or not cleanup_marker(container, EXECUTE_ABANDON_MARKER, retired_execute_key)
         ):
             raise ProviderError("retired-execute custody marker is not exact")
-        if task_command_missing and retired_execute_key is None:
+        if (
+            task_command_missing
+            and retired_execute_key is None
+            and not service_cancel_allows_missing_task_command(action)
+        ):
             raise ProviderError(
                 "missing task-command has no exact retired-execute custody proof"
             )
@@ -3236,6 +3284,11 @@ def mutate_execute(controller, action):
     if disposition in (EXECUTE_DISPOSITION_TERMINAL, EXECUTE_DISPOSITION_RECOVERED):
         if disposition == EXECUTE_DISPOSITION_RECOVERED:
             persist_execute_result(controller, action, names, tags, recovered)
+            worker = worker_by_slot(
+                inventory(controller, include_metrics=False), action["slot"]
+            )
+            if worker is None:
+                raise ProviderError("execution result persistence lost its exact worker")
         return worker, recovered
     if disposition != EXECUTE_DISPOSITION_SUBMIT:
         raise ProviderError("worker execution disposition is unsupported")
@@ -3336,7 +3389,10 @@ def mutate_execute(controller, action):
     if execution is None:
         raise ProviderError("private worker execution returned no exact result")
     persist_execute_result(controller, action, names, tags, execution)
-    return worker_by_slot(inventory(controller, include_metrics=False), action["slot"]), execution
+    worker = worker_by_slot(inventory(controller, include_metrics=False), action["slot"])
+    if worker is None:
+        raise ProviderError("execution result persistence lost its exact worker")
+    return worker, execution
 
 
 def mutate_steer(controller, action):

bin/fm-secondmate-cloud-monitor.py

@@ -235,6 +235,9 @@
 # Child model/effort ride the child's spawn argv, so the monitor is stricter
 # than the runner: a bounded shell-safe token or nothing.
 SAFE_OPTION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
+SAFE_MODEL_OPTION = re.compile(
+    r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}(?:/[A-Za-z0-9][A-Za-z0-9._-]{0,63})?$"
+)
 # The cloud lane runs exactly one runtime: fm-spawn refuses every other
 # harness under cloud placement and forces this one itself.
 CLOUD_CHILD_HARNESS = "pi"
@@ -971,9 +974,14 @@ def check_child_request(message, task, generation, assignment):
         # fm-secondmate-session.py refuses leading-dash inbox text, and the
         # same reason child_model/child_effort carry a strict charset here.
         return "request brief begins with '-' and cannot ride the pi argv"
-    for key in CHILD_REQUEST_OPTIONAL:
-        if key in message and not SAFE_OPTION.fullmatch(message[key]):
-            return "request {} is malformed".format(key)
+    if "child_model" in message and not SAFE_MODEL_OPTION.fullmatch(
+        message["child_model"]
+    ):
+        return "request child_model is malformed"
+    if "child_effort" in message and not SAFE_OPTION.fullmatch(
+        message["child_effort"]
+    ):
+        return "request child_effort is malformed"
     return ""
 
 

bin/fm-worker-lifecycle.py

@@ -2057,6 +2057,14 @@ def make_action(env, action_type, worker=None, item=None, **fields):
         })
         if worker.get("retired_execute_key") is not None:
             action["retired_execute_key"] = worker["retired_execute_key"]
+        service_cancel_proof = worker.get("release_proof")
+        if (
+            action_type == "delete-compute"
+            and isinstance(service_cancel_proof, dict)
+            and service_cancel_proof.get("schema") == "fm.worker-service-cancel/v1"
+            and service_cancel_proof.get("verdict") == "cancelled-before-execution"
+        ):
+            action["service_cancel_proof"] = copy.deepcopy(service_cancel_proof)
     if item is not None:
         action["request"] = item
     action.update(fields)
@@ -2498,6 +2506,10 @@ def apply_action_result(env, state, action, result):
             elif execution.get("step_outcome_sha256") not in (None, ""):
                 raise LifecycleError(
                     "absent no-mistakes service return asserted a step outcome digest")
+        cloud = result.get("worker")
+        if not isinstance(cloud, dict) or cloud.get("slot") != worker["slot"]:
+            raise LifecycleError("provider execute result returned the wrong worker slot")
+        adopt_cloud_resources(worker, cloud)
         state["executions"][action["request_digest"]] = execution
         worker["last_execution_digest"] = supplied
         worker["last_execution_at"] = iso_utc()

tests/fm-azure-pilot.test.sh

@@ -1479,6 +1479,51 @@ try:
     raise AssertionError("an overwrite upload took the create-once convergence path")
 except provider.ProviderError as exc:
     assert "staging upload failed" in str(exc), exc
+
+# A later execute reuses the assignment's fixed request blob name. Its action
+# can carry the ETag observed before the prior execute updated that blob, so a
+# same-assignment ConditionNotMet must adopt the current ETag and retry one CAS.
+# A foreign assignment still fails the complete expected-tag comparison.
+conditional_tags = {"task-binding": "task-one", "assignment-generation": "asg-00000001"}
+conditional_calls = []
+old_payload = b'{"old":true}\n'
+
+
+def conditional_az(controller, args, check=True, timeout=provider.AZ_TIMEOUT_SECONDS):
+    conditional_calls.append(list(args))
+    if "upload" in args:
+        match = args[args.index("--if-match") + 1]
+        if match == '"stale"':
+            return None, 1, "ErrorCode:ConditionNotMet"
+        assert match == '"current"', args
+        return {}, 0, ""
+    if "show" in args:
+        return {
+            "etag": '"current"',
+            "metadata": provider.tags_to_metadata(conditional_tags),
+            "properties": {"etag": '"current"'},
+        }, 0, ""
+    if "download" in args:
+        Path(args[args.index("--file") + 1]).write_bytes(old_payload)
+        return {}, 0, ""
+    raise AssertionError(args)
+
+
+provider.az = conditional_az
+provider.upload_json_blob(
+    controller, "acct", "worker-state-01", "request.json", {"new": True},
+    conditional_tags, overwrite=True, if_match='"stale"',
+)
+assert sum(1 for args in conditional_calls if "upload" in args) == 2, conditional_calls
+foreign_tags = dict(conditional_tags, **{"assignment-generation": "asg-00000002"})
+try:
+    provider.upload_json_blob(
+        controller, "acct", "worker-state-01", "request.json", {"newer": True},
+        foreign_tags, overwrite=True, if_match='"stale"',
+    )
+    raise AssertionError("a foreign assignment replaced the sequential execute blob")
+except provider.ProviderError as exc:
+    assert "staging upload failed" in str(exc), exc
 # The checks above drive upload_json_blob directly and so cannot see the REAL
 # caller dropping volatile_fields, which is what made S1 revertible with a
 # green suite. Capture what create_lifecycle_children actually passes.

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

@@ -1484,7 +1484,7 @@ test_chain_tip_argv_is_accepted_by_the_real_lifecycle_cli() {
 
 test_valid_child_request_spawns_with_the_exact_parent_pair() {
   make_world child-valid
-  emit_child_intent '{"kind":"ship","brief":"ship the compartment child","model":"gpt-5","effort":"high"}'
+  emit_child_intent '{"kind":"ship","brief":"ship the compartment child","model":"openai-codex/gpt-5.6-sol","effort":"high"}'
   start_monitor
   wait_for "the child spawn" test -s "$SP_LOG"
   wait_for "the acceptance delivered into the inbox" inbox_has 'FIRSTMATE ACCEPTED'
@@ -1504,7 +1504,7 @@ PY
   argv=$(sed -n 1p "$SP_LOG")
   # The project resolves out of the COMPARTMENT's home (alpha), never the
   # primary's (primary-only): the split points fm-spawn's projects/ there.
-  [ "$argv" = "$(printf '%s\x1f%s\x1f--harness\x1fpi\x1f--model\x1fgpt-5\x1f--effort\x1fhigh' "$child" "$LANDING/projects/alpha")" ] \
+  [ "$argv" = "$(printf '%s\x1f%s\x1f--harness\x1fpi\x1f--model\x1fopenai-codex/gpt-5.6-sol\x1f--effort\x1fhigh' "$child" "$LANDING/projects/alpha")" ] \
     || fail "the child spawn argv is not the exact ship shape: $(printf '%s' "$argv" | tr '\037' '|')"
   env_line=$(sed -n 2p "$SP_LOG")
   assert_contains "$env_line" "FM_HOME=$HOME_DIR" \

tests/fm-worker-lifecycle.test.sh

@@ -313,7 +313,7 @@ PY
 }
 
 service_cancel_replay_contract() {
-  python3 - "$CONTROLLER" <<'PY' || fail "service cancellation replay contract failed"
+  python3 - "$CONTROLLER" "$AZURE" <<'PY' || fail "service cancellation replay contract failed"
 import contextlib
 import copy
 import importlib.util
@@ -323,6 +323,9 @@ import sys
 spec = importlib.util.spec_from_file_location("lifecycle", sys.argv[1])
 module = importlib.util.module_from_spec(spec)
 spec.loader.exec_module(module)
+provider_spec = importlib.util.spec_from_file_location("azure_provider", sys.argv[2])
+provider = importlib.util.module_from_spec(provider_spec)
+provider_spec.loader.exec_module(provider)
 
 bindings = {
     "home_binding": "1" * 64,
@@ -368,6 +371,14 @@ module.command_service_cancel(env, args)
 assert item["status"] == "releasing", item
 assert item["service_completion_receipt"] == worker["release_proof"], item
 assert worker["release_proof"]["verdict"] == "cancelled-before-execution", worker
+action_env = {"deployment_generation": "deployment", "owner": "owner"}
+worker.update({"sku": "Standard_D4as_v6", "sku_family": "standardDav6Family", "cloud_generation": 1, "reservation_usd": 1.0})
+delete_action = module.make_action(action_env, "delete-compute", worker=worker)
+assert delete_action["service_cancel_proof"] == worker["release_proof"], delete_action
+assert provider.service_cancel_allows_missing_task_command(delete_action)
+foreign_cancel = copy.deepcopy(delete_action)
+foreign_cancel["service_cancel_proof"]["task"] = "foreign-task"
+assert not provider.service_cancel_allows_missing_task_command(foreign_cancel)
 module.command_service_cancel(env, args)
 
 item["status"] = "complete"