diff --git a/bin/fm-azure-worker-provider.py b/bin/fm-azure-worker-provider.py
index f9b90cc37cf..ebb2d0cd433 100755
--- a/bin/fm-azure-worker-provider.py
+++ b/bin/fm-azure-worker-provider.py
@@ -54,9 +54,12 @@
 RESPONSE_SCHEMA = "fm.worker-provider-response/v1"
 INVENTORY_SCHEMA = "fm.worker-provider-inventory/v1"
 EXECUTION_TERMINAL_SCHEMA = "fm.worker-execution-terminal/v1"
+EXECUTION_RESULT_SCHEMA = "fm.worker-execution-result/v1"
 EXECUTE_DISPOSITION_SUBMIT = "submit"
 EXECUTE_DISPOSITION_TERMINAL = "terminal"
 EXECUTE_DISPOSITION_RECOVERED = "recovered"
+EXECUTION_REQUEST_TAG = "execution-request-digest"
+EXECUTION_IDEMPOTENCY_TAG = "execution-idempotency-key"
 MAX_INPUT_BYTES = 2 * 1024 * 1024
 # The one blob name the guest may create, and the ceiling the supervisor
 # enforces before it uploads; both sides bound the same transfer. The name
@@ -2602,25 +2605,134 @@ def build_execute_script(action):
     )
 
 
-def execute_terminal_disposition(controller, action, task_command_resource):
+def initial_execute_staging_pair(action):
+    try:
+        supervisor_bytes = (ROOT / "bin" / "fm-worker-supervisor.py").read_bytes()
+    except OSError as exc:
+        raise ProviderError(
+            "exact initial worker supervisor is unreadable: {}".format(str(exc)[:300])
+        ) from None
+    supervisor_digest = hashlib.sha256(supervisor_bytes).hexdigest()
+    return {
+        "staging-request": {
+            "schema": "fm.worker-staging-request/v1",
+            "status": "assigned",
+            "slot": action["slot"],
+            "bindings": action["bindings"],
+            "supervisor_sha256": supervisor_digest,
+        },
+        "staging-result": {
+            "schema": "fm.worker-staging-result/v1",
+            "status": "pending",
+            "assignment_generation": action["bindings"]["assignment_generation"],
+        },
+    }
+
+
+def initial_execute_staging_is_exact(action, resources):
+    for kind, value in initial_execute_staging_pair(action).items():
+        payload = canonical_bytes(value) + b"\n"
+        resource = resources.get(kind) or {}
+        if (
+            resource.get("digest") != hashlib.sha256(payload).hexdigest()
+            or resource.get("length") != len(payload)
+        ):
+            return False
+    return True
+
+
+def run_command_execution_binding(live):
+    properties = live.get("properties") or {}
+    tags = live.get("tags") or properties.get("tags") or {}
+    request_digest = tags.get(EXECUTION_REQUEST_TAG)
+    idempotency_key = tags.get(EXECUTION_IDEMPOTENCY_TAG)
+    if request_digest is None and idempotency_key is None:
+        return None
+    if (
+        not isinstance(request_digest, str)
+        or not re.fullmatch(r"[0-9a-f]{64}", request_digest)
+        or not isinstance(idempotency_key, str)
+        or not re.fullmatch(r"[0-9a-f]{64}", idempotency_key)
+    ):
+        raise ProviderError("worker task Run Command execution binding tags are malformed")
+    return request_digest, idempotency_key
+
+
+def exact_execution_marker(view):
+    execution = marker_payload(
+        "{}\n{}".format(view.get("output", ""), view.get("error", "")),
+        "FM-WORKER-RESULT:",
+    )
+    if execution is None:
+        return None
+    if not isinstance(execution, dict) or execution.get("schema") != EXECUTION_RESULT_SCHEMA:
+        raise ProviderError("worker task Run Command result marker schema is not exact")
+    supplied = execution.get("result_digest")
+    unsigned = dict(execution)
+    unsigned.pop("result_digest", None)
+    if supplied != hashlib.sha256(canonical_bytes(unsigned)).hexdigest():
+        raise ProviderError("recovered private worker result digest is not exact")
+    return execution
+
+
+def execute_terminal_disposition(controller, action, resources):
+    task_command_resource = resources.get("task-command") if isinstance(resources, dict) else None
     if not isinstance(task_command_resource, dict) or not task_command_resource.get("id"):
         return EXECUTE_DISPOSITION_SUBMIT, None
     live = show_full(controller, task_command_resource["id"])
     properties = live.get("properties") or {}
     source = properties.get("source") or live.get("source") or {}
     stored_script = source.get("script") if isinstance(source, dict) else None
-    if not isinstance(stored_script, str) or not stored_script.strip():
-        raise ProviderError("existing worker task Run Command source script is unreadable")
     expected_script = build_execute_script(action)
     request_digest = action.get("request_digest")
-    exact_script = stored_script == expected_script
+    idempotency_key = action.get("idempotency_key")
+    binding = run_command_execution_binding(live)
+    current_binding = (request_digest, idempotency_key)
+    substantive_script = isinstance(stored_script, str) and bool(stored_script.strip())
+    exact_script = substantive_script and stored_script == expected_script
     fallback_bound = (
-        isinstance(request_digest, str)
+        substantive_script
+        and isinstance(request_digest, str)
         and re.fullmatch(r"[0-9a-f]{64}", request_digest)
         and request_digest in stored_script
         and execute_generation_line(action) in stored_script.splitlines()
     )
-    if not exact_script and not fallback_bound:
+    view = None
+    if binding is not None and binding != current_binding:
+        if exact_script or fallback_bound:
+            raise ProviderError("worker task Run Command script and execution binding tags disagree")
+        view = run_command_instance_view(
+            controller, expected_names(controller, action["slot"])["vm"],
+            expected_names(controller, action["slot"])["task-command"],
+        )
+        previous = exact_execution_marker(view)
+        if (
+            view.get("executionState") == "Succeeded"
+            and isinstance(previous, dict)
+            and previous.get("request_digest") == binding[0]
+        ):
+            return EXECUTE_DISPOSITION_SUBMIT, None
+        raise ProviderError("worker task Run Command has an ambiguous prior execution binding")
+    if not substantive_script:
+        if binding is None:
+            if not initial_execute_staging_is_exact(action, resources):
+                raise ProviderError(
+                    "existing worker task Run Command source is unreadable outside the exact initial staging state"
+                )
+            view = run_command_instance_view(
+                controller, expected_names(controller, action["slot"])["vm"],
+                expected_names(controller, action["slot"])["task-command"],
+            )
+            if (
+                view.get("executionState") == "Failed"
+                and str(view.get("exitCode")) == "-202"
+                and exact_execution_marker(view) is None
+            ):
+                return EXECUTE_DISPOSITION_SUBMIT, None
+            raise ProviderError("worker task Run Command does not prove the exact initial preflight stub")
+    elif not exact_script and not fallback_bound:
+        if binding == current_binding:
+            raise ProviderError("worker task Run Command source disagrees with its exact execution binding")
         return EXECUTE_DISPOSITION_SUBMIT, None
     provisioning_state = properties.get("provisioningState") or live.get("provisioningState")
     if str(provisioning_state).lower() in ("failed", "canceled"):
@@ -2639,24 +2751,16 @@ def execute_terminal_disposition(controller, action, task_command_resource):
             )
         )
     names = expected_names(controller, action["slot"])
-    view = run_command_instance_view(controller, names["vm"], names["task-command"])
+    view = view or run_command_instance_view(controller, names["vm"], names["task-command"])
     if view.get("executionState") != "Succeeded":
         raise ProviderError(
             "exact worker execution has no recoverable terminal result: state={}".format(
                 view.get("executionState")
             )
         )
-    execution = marker_payload(
-        "{}\n{}".format(view.get("output", ""), view.get("error", "")),
-        "FM-WORKER-RESULT:",
-    )
+    execution = exact_execution_marker(view)
     if not isinstance(execution, dict) or execution.get("request_digest") != request_digest:
         raise ProviderError("exact worker execution has no request-bound result marker")
-    supplied = execution.get("result_digest")
-    unsigned = dict(execution)
-    unsigned.pop("result_digest", None)
-    if supplied != hashlib.sha256(canonical_bytes(unsigned)).hexdigest():
-        raise ProviderError("recovered private worker result digest is not exact")
     return EXECUTE_DISPOSITION_RECOVERED, execution
 
 
@@ -2703,7 +2807,7 @@ def mutate_execute(controller, action):
     names = expected_names(controller, action["slot"])
     tags = action_tags(controller, action)
     disposition, recovered = execute_terminal_disposition(
-        controller, action, resources.get("task-command")
+        controller, action, resources
     )
     if disposition in (EXECUTE_DISPOSITION_TERMINAL, EXECUTE_DISPOSITION_RECOVERED):
         if disposition == EXECUTE_DISPOSITION_RECOVERED:
@@ -2721,9 +2825,9 @@ def mutate_execute(controller, action):
     # the archives ride private blobs and reach the guest over short-lived
     # read-only user-delegation SAS bounded to the wall plus collection slack.
     # The SAS URLs travel as PROTECTED run-command parameters: ARM GET and
-    # az vm run-command show return source.script but never protected
-    # parameters, so the account-archive SAS is not readable off the control
-    # plane for its validity window. The managed run-command agent delivers
+    # az vm run-command show expose the ordinary resource and binding tags but
+    # never protected parameters, so the account-archive SAS is not readable
+    # off the control plane for its validity window. The managed agent delivers
     # parameters as environment variables for the script process, which the
     # supervisor inherits.
     protected_parameters = []
@@ -2766,6 +2870,11 @@ def mutate_execute(controller, action):
             raise ProviderError("outcome staging SAS carries unsupported characters")
         protected_parameters.append("FM_WORKER_OUTCOME_URL=" + outcome_sas)
     script = build_execute_script(action)
+    execution_tags = dict(tags)
+    execution_tags.update({
+        EXECUTION_REQUEST_TAG: action["request_digest"],
+        EXECUTION_IDEMPOTENCY_TAG: action["idempotency_key"],
+    })
     update_command = [
         "vm", "run-command", "update", "--resource-group", controller["resource_group"],
         "--vm-name", names["vm"], "--name", names["task-command"],
@@ -2775,6 +2884,9 @@ def mutate_execute(controller, action):
         # the client still blocked. bin/fm-azure-runner.py and
         # bin/fm-azure-validation.py both set it for the same reason.
         "--timeout-in-seconds", str(int(request["wall_seconds"]) + GUEST_RUN_SLACK_SECONDS),
+        "--tags",
+    ] + [
+        "{}={}".format(key, value) for key, value in sorted(execution_tags.items())
     ]
     if protected_parameters:
         update_command += ["--protected-parameters"] + protected_parameters
@@ -2795,16 +2907,9 @@ def mutate_execute(controller, action):
         raise ProviderError("private worker execution did not complete in the guest: state={} error={}".format(
             view.get("executionState"), str(view.get("error", ""))[:500]
         ))
-    execution = marker_payload(
-        "{}\n{}".format(view.get("output", ""), view.get("error", "")), "FM-WORKER-RESULT:"
-    )
+    execution = exact_execution_marker(view)
     if execution is None:
         raise ProviderError("private worker execution returned no exact result")
-    supplied = execution.get("result_digest")
-    unsigned = dict(execution)
-    unsigned.pop("result_digest", None)
-    if supplied != hashlib.sha256(canonical_bytes(unsigned)).hexdigest():
-        raise ProviderError("private worker result digest is not exact")
     persist_execute_result(controller, action, names, tags, execution)
     return worker_by_slot(inventory(controller, include_metrics=False), action["slot"]), execution
 
diff --git a/docs/azure-workers.md b/docs/azure-workers.md
index 6d472ff7459..0bb63c9a2a8 100644
--- a/docs/azure-workers.md
+++ b/docs/azure-workers.md
@@ -328,7 +328,7 @@ The home lock now covers only short read-validate-claim and apply sections; ever
 Reconcile drains stranded claims AFTER convergence, skipping any slot whose claim a live process still owns, so a wedged or hours-long replay cannot stop the fleet.
 `abandon-claim` can record and clear either an exact-key provider result whose apply deterministically refuses, including a script-bound Failed or Canceled execution disposition that names the claimed task-command resource, or an exact-key `REFUSED-IDENTITY` replay whose recorded resource identity can never bind again.
 Both dispositions are recorded in `cleanup_refusals` before the claim is cleared, while an ordinary transient provider failure retains the claim unchanged.
-An existing Run Command whose source script is missing, non-string, empty, or whitespace-only cannot prove that a first submission is safe and retains the claim. An exact-bound Run Command that is still Updating or Running, reports Succeeded without a digest-valid request-bound result marker, or returns a terminal disposition naming a foreign task-command resource likewise retains the claim without submitting the command again.
+Azure may omit `source.script` even for the freshly created async preflight stub. That shape permits its one initial submission only while both staging blobs still carry the exact assignment/pending sentinels and instance view is Failed with exit `-202` and no result marker. The execution update atomically tags the Run Command with its request digest and idempotency key; a replay bearing those tags is bound even when source remains absent. A crash after staging changes but before the tagged update stays fail-closed. Every other missing, non-string, empty, or whitespace-only source is ambiguous unless a digest-valid result proves a differently tagged prior execution completed. An exact-bound Run Command that is still Updating or Running, reports Succeeded without a digest-valid request-bound result marker, or returns a terminal disposition naming a foreign task-command resource likewise retains the claim without submitting the command again.
 Each reconcile refreshes Azure before selecting the next action and stops after 64 actions even if a provider never converges.
 A provider error preserves the slot's pending action and records a bounded cleanup refusal.
 The next controller process replays that exact action before considering new work.
diff --git a/tests/fm-worker-lifecycle.test.sh b/tests/fm-worker-lifecycle.test.sh
index b7110d0a0a8..d03c75c8c4e 100755
--- a/tests/fm-worker-lifecycle.test.sh
+++ b/tests/fm-worker-lifecycle.test.sh
@@ -569,7 +569,7 @@ module.show_full = lambda *_args, **_kwargs: {
     "properties": {"source": {"script": script}, "provisioningState": "Failed"}
 }
 terminal_kind, terminal = module.execute_terminal_disposition(
-    controller, execute_action, task_command
+    controller, execute_action, worker["resources"]
 )
 assert terminal_kind == module.EXECUTE_DISPOSITION_TERMINAL, terminal_kind
 assert terminal == {
@@ -585,7 +585,7 @@ module.show_full = lambda *_args, **_kwargs: {
                    "provisioningState": "Canceled"}
 }
 fallback_kind, fallback_terminal = module.execute_terminal_disposition(
-    controller, execute_action, task_command
+    controller, execute_action, worker["resources"]
 )
 assert fallback_kind == module.EXECUTE_DISPOSITION_TERMINAL, fallback_kind
 assert fallback_terminal["disposition"] == "provider-terminal", fallback_terminal
@@ -593,13 +593,13 @@ stale_script = script.replace(execute_action["request_digest"], "9" * 64)
 module.show_full = lambda *_args, **_kwargs: {
     "properties": {"source": {"script": stale_script}, "provisioningState": "Failed"}
 }
-assert module.execute_terminal_disposition(controller, execute_action, task_command) == (
+assert module.execute_terminal_disposition(controller, execute_action, worker["resources"]) == (
     module.EXECUTE_DISPOSITION_SUBMIT, None
 )
 module.show_full = lambda *_args, **_kwargs: {
     "properties": {"source": {"script": "echo unrelated"}, "provisioningState": "Failed"}
 }
-assert module.execute_terminal_disposition(controller, execute_action, task_command) == (
+assert module.execute_terminal_disposition(controller, execute_action, worker["resources"]) == (
     module.EXECUTE_DISPOSITION_SUBMIT, None
 )
 execution = {
@@ -622,7 +622,7 @@ module.run_command_instance_view = lambda *_args, **_kwargs: {
     "error": "",
 }
 recovered_kind, recovered_execution = module.execute_terminal_disposition(
-    controller, execute_action, task_command
+    controller, execute_action, worker["resources"]
 )
 assert recovered_kind == module.EXECUTE_DISPOSITION_RECOVERED, recovered_kind
 assert recovered_execution == execution, recovered_execution
@@ -638,7 +638,8 @@ real_upload_json_blob = module.upload_json_blob
 real_az = module.az
 module.inventory = lambda *_args, **_kwargs: {"workers": [worker]}
 module.worker_by_slot = lambda _snapshot, _slot: worker
-module.recorded_exact = lambda _action, _worker: worker["resources"]
+active_resources = {"value": worker["resources"]}
+module.recorded_exact = lambda _action, _worker: active_resources["value"]
 updates = []
 def forbidden_update(*_args, **_kwargs):
     updates.append("update")
@@ -661,6 +662,58 @@ retained_execute({"provisioningState": "Succeeded"})
 retained_execute({"source": {"script": None}, "provisioningState": "Succeeded"})
 retained_execute({"source": {"script": ""}, "provisioningState": "Succeeded"})
 retained_execute({"source": {"script": " \n\t"}, "provisioningState": "Succeeded"})
+retained_execute(
+    {
+        "tags": {
+            module.EXECUTION_REQUEST_TAG: execute_action["request_digest"],
+            module.EXECUTION_IDEMPOTENCY_TAG: execute_action["idempotency_key"],
+        },
+        "provisioningState": "Succeeded",
+    },
+    {"executionState": "Succeeded", "output": "", "error": ""},
+)
+retained_execute(
+    {
+        "source": {"script": "echo prior execution"},
+        "tags": {
+            module.EXECUTION_REQUEST_TAG: "9" * 64,
+            module.EXECUTION_IDEMPOTENCY_TAG: "8" * 64,
+        },
+        "provisioningState": "Succeeded",
+    },
+    {"executionState": "Succeeded", "output": "", "error": ""},
+)
+prior_execution = dict(execution, request_digest="9" * 64)
+prior_execution.pop("result_digest", None)
+prior_execution["result_digest"] = hashlib.sha256(
+    module.canonical_bytes(prior_execution)
+).hexdigest()
+module.show_full = lambda *_args, **_kwargs: {
+    "properties": {
+        "source": {"script": "echo prior execution"},
+        "provisioningState": "Succeeded",
+    },
+    "tags": {
+        module.EXECUTION_REQUEST_TAG: "9" * 64,
+        module.EXECUTION_IDEMPOTENCY_TAG: "8" * 64,
+    },
+}
+module.run_command_instance_view = lambda *_args, **_kwargs: {
+    "executionState": "Succeeded",
+    "output": "FM-WORKER-RESULT:" + json.dumps(
+        prior_execution, sort_keys=True, separators=(",", ":")
+    ),
+    "error": "",
+}
+assert module.execute_terminal_disposition(
+    controller, execute_action, active_resources["value"]
+) == (module.EXECUTE_DISPOSITION_SUBMIT, None)
+retained_execute(
+    {
+        "tags": {module.EXECUTION_REQUEST_TAG: execute_action["request_digest"]},
+        "provisioningState": "Succeeded",
+    }
+)
 retained_execute({"source": {"script": script}, "provisioningState": "Updating"})
 retained_execute(
     {"source": {"script": script}, "provisioningState": "Succeeded"},
@@ -681,6 +734,137 @@ retained_execute(
         "error": "",
     },
 )
+
+# The exact initial sentinel depends on the same pinned supervisor bytes used
+# at worker creation. A missing or unreadable local copy is a bounded provider
+# refusal, never an unhandled filesystem exception.
+real_root = module.ROOT
+class UnreadableRoot:
+    def __truediv__(self, _component):
+        return self
+
+    def read_bytes(self):
+        raise OSError("unreadable-" + "x" * 1000)
+
+module.ROOT = UnreadableRoot()
+try:
+    module.initial_execute_staging_pair(execute_action)
+except module.ProviderError as exc:
+    assert str(exc).startswith("exact initial worker supervisor is unreadable: "), exc
+    assert len(str(exc)) <= 350, len(str(exc))
+else:
+    raise AssertionError("unreadable worker supervisor escaped the provider refusal boundary")
+finally:
+    module.ROOT = real_root
+
+# Azure's actual fresh managed Run Command omits source entirely and exposes
+# the async preflight stub as Failed/-202. The exact untouched staging pair is
+# the run-owned proof that this is the one initial submission. Any other exit
+# remains ambiguous. The accepted update atomically carries both execution
+# bindings in its tags, so a replay cannot pass through this initial gate.
+fresh_resources = copy.deepcopy(worker["resources"])
+for kind, value in module.initial_execute_staging_pair(execute_action).items():
+    body = module.canonical_bytes(value) + b"\n"
+    fresh_resources[kind]["digest"] = hashlib.sha256(body).hexdigest()
+    fresh_resources[kind]["length"] = len(body)
+crashed_resources = copy.deepcopy(fresh_resources)
+crashed_resources["staging-request"]["digest"] = "0" * 64
+active_resources["value"] = crashed_resources
+retained_execute(
+    {"provisioningState": "Succeeded"},
+    {"executionState": "Failed", "exitCode": -202, "output": "", "error": ""},
+)
+active_resources["value"] = fresh_resources
+retained_execute(
+    {"provisioningState": "Succeeded"},
+    {"executionState": "Failed", "exitCode": -201, "output": "", "error": ""},
+)
+
+uploaded_json_blobs = []
+def record_json_upload(_controller, _storage, _container, blob_name, *_args, **_kwargs):
+    uploaded_json_blobs.append(blob_name)
+module.upload_json_blob = record_json_upload
+views = iter([
+    {"executionState": "Failed", "exitCode": -202, "output": "", "error": ""},
+    {
+        "executionState": "Succeeded", "exitCode": 0,
+        "output": "FM-WORKER-RESULT:" + json.dumps(
+            execution, sort_keys=True, separators=(",", ":")
+        ),
+        "error": "",
+    },
+])
+module.show_full = lambda *_args, **_kwargs: {
+    "properties": {"provisioningState": "Succeeded"},
+    "tags": dict(tags),
+}
+update_commands = []
+def accept_initial_update(_controller, command, **_kwargs):
+    update_commands.append(command)
+    return {}, 0, ""
+module.az = accept_initial_update
+
+# A just-submitted command is held to the same exact schema and digest parser
+# used during recovery. A self-digested foreign schema must not be persisted.
+wrong_schema_execution = dict(execution, schema="fm.worker-execution-result/v2")
+wrong_schema_execution.pop("result_digest", None)
+wrong_schema_execution["result_digest"] = hashlib.sha256(
+    module.canonical_bytes(wrong_schema_execution)
+).hexdigest()
+wrong_schema_views = iter([
+    {"executionState": "Failed", "exitCode": -202, "output": "", "error": ""},
+    {
+        "executionState": "Succeeded", "exitCode": 0,
+        "output": "FM-WORKER-RESULT:" + json.dumps(
+            wrong_schema_execution, sort_keys=True, separators=(",", ":")
+        ),
+        "error": "",
+    },
+])
+module.run_command_instance_view = lambda *_args, **_kwargs: next(wrong_schema_views)
+try:
+    module.mutate_execute(controller, execute_action)
+except module.ProviderError as exc:
+    assert "result marker schema is not exact" in str(exc), exc
+else:
+    raise AssertionError("wrong-schema submitted execution result was accepted")
+assert len(update_commands) == 1, update_commands
+assert uploaded_json_blobs == [
+    module.expected_names(controller, execute_action["slot"])["staging-request"]
+], uploaded_json_blobs
+
+update_commands.clear()
+uploaded_json_blobs.clear()
+module.run_command_instance_view = lambda *_args, **_kwargs: next(views)
+_, admitted_execution = module.mutate_execute(controller, execute_action)
+assert admitted_execution == execution, admitted_execution
+assert len(update_commands) == 1, update_commands
+assert uploaded_json_blobs == [
+    module.expected_names(controller, execute_action["slot"])["staging-request"],
+    module.expected_names(controller, execute_action["slot"])["staging-result"],
+], uploaded_json_blobs
+submitted = update_commands[0]
+assert "{}={}".format(
+    module.EXECUTION_REQUEST_TAG, execute_action["request_digest"]
+) in submitted, submitted
+assert "{}={}".format(
+    module.EXECUTION_IDEMPOTENCY_TAG, execute_action["idempotency_key"]
+) in submitted, submitted
+
+updates.clear()
+module.upload_json_blob = forbidden_update
+module.az = forbidden_update
+retained_execute(
+    {
+        "tags": {
+            module.EXECUTION_REQUEST_TAG: execute_action["request_digest"],
+            module.EXECUTION_IDEMPOTENCY_TAG: execute_action["idempotency_key"],
+        },
+        "provisioningState": "Succeeded",
+    },
+    {"executionState": "Succeeded", "output": "", "error": ""},
+)
+active_resources["value"] = worker["resources"]
 module.inventory = real_inventory
 module.worker_by_slot = real_worker_by_slot
 module.recorded_exact = real_recorded_exact
@@ -1071,7 +1255,7 @@ if path.exists():
 else:
     state = {
         "workers": {}, "seen": {}, "calls": [], "execute_updates": 0,
-        "execute_terminal_probes": 0,
+        "execute_terminal_probes": 0, "initial_stub_admissions": 0,
         "metrics": {"actual_usd": 100.0, "forecast_usd": 150.0},
     }
 
@@ -1182,6 +1366,11 @@ if request["operation"] == "mutate":
                 )
                 raise SystemExit(3)
     task_command = (live_worker or {}).get("resources", {}).get("task-command", {})
+    if kind == "execute" and os.environ.get("FIXTURE_INITIAL_EXECUTE_STUB"):
+        if task_command.get("execution_request_digest") or task_command.get("execution_idempotency_key"):
+            sys.stderr.write("FIXTURE PROVIDER REFUSED: initial execute stub is already bound\n")
+            raise SystemExit(2)
+        state["initial_stub_admissions"] = state.get("initial_stub_admissions", 0) + 1
     if kind == "execute" and os.environ.get("FIXTURE_UNREADABLE_EXECUTE_SCRIPT"):
         state["execute_terminal_probes"] = state.get("execute_terminal_probes", 0) + 1
         save()
@@ -1278,6 +1467,8 @@ if request["operation"] == "mutate":
             task_command["immutable_id"] = "task-command-execute-{}".format(key[:8])
             task_command["provisioning_state"] = "Succeeded"
             task_command["request_digest"] = action["request_digest"]
+            task_command["execution_request_digest"] = action["request_digest"]
+            task_command["execution_idempotency_key"] = key
             execution = {
                 "schema": "fm.worker-execution-result/v1",
                 "request_digest": action["request_digest"],
@@ -1557,13 +1748,18 @@ old_assignment = state["workers"][str(old_slot)]["assignment_generation"]
 
 # The private one-task execution path returns an exact result and replays the
 # same request from durable state without a second provider execution.
-execution = json.loads(run(
-    "execute", "--task", "task-1", "--task-generation", "gen-1",
-    "--assignment-generation", old_assignment, "--wall-seconds", "60",
-    "--confirm-execute", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"],
-    "--", "/usr/bin/true",
-).stdout)
+initial_stub_run = subprocess.run(
+    [wrapper, "execute", "--task", "task-1", "--task-generation", "gen-1",
+     "--assignment-generation", old_assignment, "--wall-seconds", "60",
+     "--confirm-execute", "--confirm-subscription", env["FM_AZURE_SUBSCRIPTION_ID"],
+     "--", "/usr/bin/true"],
+    env=dict(env, FIXTURE_INITIAL_EXECUTE_STUB="1"), text=True,
+    stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+)
+assert initial_stub_run.returncode == 0, initial_stub_run.stderr
+execution = json.loads(initial_stub_run.stdout)
 assert execution["schema"] == "fm.worker-execution-result/v1" and execution["exit_code"] == 0
+assert fixture_state().get("initial_stub_admissions") == 1, fixture_state()
 call_count = len(fixture_state()["calls"])
 repeat = json.loads(run(
     "execute", "--task", "task-1", "--task-generation", "gen-1",