fix(k1): fence start endpoint handoff

src/k1link/device_plugins/xgrids_k1/connection_supervisor.py

@@ -985,6 +985,69 @@ class ConnectionSupervisor:
             self._revision += 1
             return True
 
+    def endpoint_observation_has_remaining_lease(
+        self,
+        *,
+        target: EndpointTarget,
+        intent_id: str,
+        host_path_epoch: int,
+        minimum_remaining_seconds: float,
+    ) -> bool:
+        """Return whether an exact reachable endpoint outlives a command window.
+
+        A fact can still be nominally reachable while being too close to the
+        transport silence TTL to survive durable command preparation.  This
+        read-only predicate uses the same monotonic and suspend-aware clocks as
+        expiry; wall-clock timestamps and browser polling cannot extend it.
+        """
+
+        if not _nonblank(intent_id):
+            raise ConnectionSupervisorError(
+                "endpoint lease check requires an exact intent"
+            )
+        if (
+            not isinstance(minimum_remaining_seconds, (int, float))
+            or isinstance(minimum_remaining_seconds, bool)
+            or not math.isfinite(float(minimum_remaining_seconds))
+            or float(minimum_remaining_seconds) < 0.0
+        ):
+            raise ConnectionSupervisorError(
+                "endpoint minimum remaining lease is invalid"
+            )
+        with self._lock:
+            self._require_open_locked()
+            if not (
+                self._intent is not None
+                and self._intent.intent_id == intent_id
+                and self._host_path.available
+                and self._host_path.epoch == host_path_epoch
+                and self._endpoint.target == target
+                and self._endpoint.intent_id == intent_id
+                and self._endpoint.host_path_epoch == host_path_epoch
+                and self._endpoint.tcp_state == "reachable"
+                and self._endpoint_observed_monotonic is not None
+                and self._endpoint_observed_suspend_aware is not None
+            ):
+                return False
+            monotonic_elapsed = (
+                self._monotonic_now() - self._endpoint_observed_monotonic
+            )
+            suspend_aware_elapsed = (
+                self._suspend_aware_now()
+                - self._endpoint_observed_suspend_aware
+            )
+            maximum_age = max(monotonic_elapsed, suspend_aware_elapsed)
+            required_remaining = min(
+                float(minimum_remaining_seconds),
+                self._observation_ttl_seconds,
+            )
+            return bool(
+                monotonic_elapsed >= 0.0
+                and suspend_aware_elapsed >= 0.0
+                and maximum_age + required_remaining
+                <= self._observation_ttl_seconds
+            )
+
     def refresh_control_evidence(
         self,
         *,

src/k1link/device_plugins/xgrids_k1/facade.py

@@ -290,6 +290,12 @@ CONNECTION_MONITOR_ASSOCIATION_TIMEOUT_SECONDS = 3.0
 # A technical observer timeout is still handled below only by the exact
 # kernel-route/TCP/control-proof retention gate.
 COMMAND_BOUND_ASSOCIATION_TIMEOUT_SECONDS = 8.0
+# Leave enough endpoint lease for durable PREPARED + worker handoff.  A field
+# run reached START with about four seconds remaining: the synchronous guard
+# passed, then the exact same endpoint expired before the worker publish guard.
+# Twelve seconds covers that observed seam without extending either TTL or
+# granting authority from cached TCP evidence.
+COMMAND_ENDPOINT_MIN_REMAINING_SECONDS = 12.0
 CONNECTION_MONITOR_HOST_CONTACT_BOUND_SECONDS = (
     HOST_ROUTE_INSPECTION_TIMEOUT_SECONDS + CONNECTION_MONITOR_ASSOCIATION_TIMEOUT_SECONDS
 )
@@ -3172,6 +3178,13 @@ class XgridsK1CompatibilityService:
                 raise ApplicationConnectionBindingLost(
                     "настройка сети K1 уже началась; прежний control path отозван"
                 )
+            acquisition = self._acquisition
+            start_dispatch_window = bool(
+                acquisition is not None
+                and acquisition.control_mode == "plugin-commanded"
+                and acquisition.state in {"starting", "awaiting_external_start"}
+                and self._acquisition_start_operation_id is not None
+            )
         target = EndpointTarget(binding.target_ipv4, binding.target_port)
         path = self._sample_host_path(
             binding.target_ipv4,
@@ -3188,6 +3201,15 @@ class XgridsK1CompatibilityService:
                 "маршрут K1 изменился во время контрольной проверки"
             )
         supervisor = self._connection_supervisor.snapshot()
+        endpoint_lease_covers_dispatch = bool(
+            not start_dispatch_window
+            or self._connection_supervisor.endpoint_observation_has_remaining_lease(
+                target=target,
+                intent_id=binding.intent_id,
+                host_path_epoch=binding.host_path_epoch,
+                minimum_remaining_seconds=COMMAND_ENDPOINT_MIN_REMAINING_SECONDS,
+            )
+        )
         if (
             path.available
             and path.route_class == "direct"
@@ -3196,15 +3218,21 @@ class XgridsK1CompatibilityService:
             and supervisor.endpoint.target == target
             and supervisor.endpoint.intent_id == binding.intent_id
             and supervisor.endpoint.host_path_epoch == binding.host_path_epoch
-            and supervisor.endpoint.reason_code == "endpoint-observation-stale"
+            and (
+                supervisor.endpoint.reason_code == "endpoint-observation-stale"
+                or (
+                    start_dispatch_window
+                    and supervisor.endpoint.tcp_state == "reachable"
+                    and not endpoint_lease_covers_dispatch
+                )
+            )
         ):
-            # Project preparation can legitimately outlive the supervisor's
+            # Project preparation can outlive the supervisor's TCP silence
-            # TCP silence lease while the exact MQTT control proof remains
+            # lease, or leave too little lease for durable PREPARED plus the
-            # fresh.  Refresh only that expired read-only endpoint fact before
+            # worker publish guard. Refresh only that exact read-only endpoint
-            # the physical START admission.  This is not a command retry: no
+            # fact. This is not a command retry. A second route sample binds
-            # operation or physical ledger row exists yet.  A second route
+            # the TCP result to the same host/association epoch; any changed
-            # sample binds the TCP result to the same host/association epoch;
+            # path or unreachable broker still fails closed.
-            # any changed path or unreachable broker still fails closed.
             endpoint = _probe_control_endpoint_socket(binding.target_ipv4)
             final_path = self._sample_host_path(
                 binding.target_ipv4,
@@ -6742,7 +6770,15 @@ class XgridsK1CompatibilityService:
                 if prepared_stop_acquisition is not None
                 else None
             )
+            prepared_start_operation_id = self._acquisition_start_operation_id
             prepared_stop_operation_id = self._acquisition_stop_operation_id
+        prepared_start_worker_outcome = self._settle_prepared_start_worker_failure(
+            acquisition_id=prepared_stop_acquisition_id,
+            start_operation_id=prepared_start_operation_id,
+        )
+        # START zero-dispatch settlement changes the durable head before the
+        # generic reducer classifies local runtime ownership.
+        physical_command_proof = self._physical_command_coordinator.snapshot()
         # Settle an exact PREPARED/no-publish STOP while the failed worker is
         # still the current control owner.  Proven-loss teardown below may
         # synchronously retire that owner to ``idle``; doing this afterwards
@@ -6767,6 +6803,7 @@ class XgridsK1CompatibilityService:
             application_control_session,
             terminal_control_proof=terminal_control_proof,
             physical_command_proof=physical_command_proof,
+            prepared_start_worker_outcome=prepared_start_worker_outcome,
             prepared_stop_worker_outcome=prepared_stop_worker_outcome,
         )
         # Acquisition reconciliation may locally retire an expired STOP
@@ -13086,6 +13123,24 @@ class XgridsK1CompatibilityService:
             self._set_active_acquisition_checkpoint_reason(None)
             return
 
+        self._cease_prepared_start_checkpoint_not_dispatched(
+            operation_id=context.operation_id,
+            acquisition_id=context.acquisition_id,
+            payload_sha256=envelope.payload_sha256,
+        )
+
+    def _cease_prepared_start_checkpoint_not_dispatched(
+        self,
+        *,
+        operation_id: str,
+        acquisition_id: str,
+        payload_sha256: str,
+    ) -> None:
+        """Cease the exact START checkpoint after durable zero-dispatch."""
+
+        store = self._active_acquisition_checkpoint
+        if store is None:
+            return
         checkpoint_snapshot = store.snapshot()
         checkpoint = checkpoint_snapshot.checkpoint
         ledger_snapshot = self._physical_command_ledger.snapshot()
@@ -13093,10 +13148,10 @@ class XgridsK1CompatibilityService:
         if not (
             ledger_snapshot.status == "resolved"
             and record is not None
-            and record.operation_id == context.operation_id
+            and record.operation_id == operation_id
-            and record.acquisition_id == context.acquisition_id
+            and record.acquisition_id == acquisition_id
             and record.action == "start"
-            and record.payload_sha256 == envelope.payload_sha256
+            and record.payload_sha256 == payload_sha256
             and record.resolution == "not-dispatched"
         ):
             raise ActiveAcquisitionRecoveryCheckpointError(
@@ -13116,9 +13171,9 @@ class XgridsK1CompatibilityService:
             return
         if not (
             checkpoint.state == "prepared"
-            and checkpoint.acquisition_id == context.acquisition_id
+            and checkpoint.acquisition_id == acquisition_id
-            and checkpoint.original_start_operation_id == context.operation_id
+            and checkpoint.original_start_operation_id == operation_id
-            and checkpoint.start_payload_sha256 == envelope.payload_sha256
+            and checkpoint.start_payload_sha256 == payload_sha256
         ):
             raise ActiveAcquisitionRecoveryCheckpointError(
                 "failed START does not own the current prepared checkpoint"
@@ -27532,6 +27587,95 @@ class XgridsK1CompatibilityService:
             },
         )
 
+    def _settle_prepared_start_worker_failure(
+        self,
+        *,
+        acquisition_id: str | None,
+        start_operation_id: str | None,
+    ) -> Literal["settled", "defer", "settlement-failed", "not-applicable"]:
+        """Seal an exact terminal START failure that never reached publish."""
+
+        if not isinstance(acquisition_id, str) or not isinstance(
+            start_operation_id, str
+        ):
+            return "not-applicable"
+
+        def matching_prepared_start(
+            physical: Mapping[str, object],
+        ) -> Mapping[str, object] | None:
+            record = physical.get("record")
+            if not (
+                physical.get("status") == "unresolved"
+                and isinstance(record, Mapping)
+                and record.get("operation_id") == start_operation_id
+                and record.get("acquisition_id") == acquisition_id
+                and record.get("action") == "start"
+                and record.get("stage") == "prepared"
+                and record.get("resolution") is None
+                and record.get("publish_call_returned") is None
+                and record.get("packet_id") is None
+                and record.get("qos2_completed") is False
+                and record.get("application_response") is None
+            ):
+                return None
+            return record
+
+        if matching_prepared_start(
+            self._physical_command_coordinator.snapshot()
+        ) is None:
+            return "not-applicable"
+        if not self._k1_command_dispatch_gate.acquire(blocking=False):
+            return "defer"
+        try:
+            current_control = dict(self._application_control_session.snapshot())
+            current_physical = dict(self._physical_command_coordinator.snapshot())
+            record = matching_prepared_start(current_physical)
+            if record is None:
+                return "not-applicable"
+            failure = current_control.get("failure")
+            if not (
+                current_control.get("state") in {"failed", "closed"}
+                and current_control.get("outcome_unknown") is False
+                and isinstance(failure, Mapping)
+                and failure.get("failed_phase") in {"start-requested", "initializing"}
+                and failure.get("modeling_command_attempted") is False
+                and failure.get("diagnostic_evidence_unavailable") == []
+            ):
+                return "not-applicable"
+            payload_sha256 = record.get("payload_sha256")
+            if not isinstance(payload_sha256, str):
+                return "not-applicable"
+            try:
+                self._physical_command_coordinator.resolve_prepared_not_dispatched(
+                    "start"
+                )
+                self._cease_prepared_start_checkpoint_not_dispatched(
+                    operation_id=start_operation_id,
+                    acquisition_id=acquisition_id,
+                    payload_sha256=payload_sha256,
+                )
+            except (ActiveAcquisitionRecoveryCheckpointError, OSError, ValueError) as exc:
+                reason_code = str(
+                    getattr(
+                        exc,
+                        "reason_code",
+                        "prepared-start-checkpoint-settlement-failed",
+                    )
+                )
+                self._set_active_acquisition_checkpoint_reason(reason_code)
+                logger.exception(
+                    "zero-publish START checkpoint settlement failed closed",
+                    extra={
+                        "event_code": "k1_start_not_dispatched_settlement_failed",
+                        "reason_code": reason_code,
+                        "device_command_retried": False,
+                    },
+                )
+                return "settlement-failed"
+            return "settled"
+        finally:
+            self._k1_command_dispatch_gate.release()
+
     def _settle_prepared_stop_worker_failure(
         self,
         *,
@@ -27816,6 +27960,9 @@ class XgridsK1CompatibilityService:
         *,
         terminal_control_proof: Mapping[str, Any] | None = None,
         physical_command_proof: Mapping[str, Any] | None = None,
+        prepared_start_worker_outcome: Literal[
+            "settled", "defer", "settlement-failed", "not-applicable"
+        ] = "not-applicable",
         prepared_stop_worker_outcome: Literal[
             "settled", "defer", "prepared-pending", "not-applicable"
         ] = "not-applicable",
@@ -28273,6 +28420,7 @@ class XgridsK1CompatibilityService:
         completed_operation_id: str | None = None
         failed_operation_id: str | None = None
         failed_start_physical_side_effect: Literal["unknown", "succeeded"] | None = None
+        zero_dispatch_start_operation_id: str | None = None
         unknown_start_operation_id: str | None = None
         unknown_start_side_effect_status: Literal["unknown", "succeeded"] | None = None
         failed_stop_operation_id: str | None = None
@@ -28584,6 +28732,28 @@ class XgridsK1CompatibilityService:
                 # Explicit STOP changes the acquisition state first and is
                 # intentionally outside this fence.
                 pass
+            elif (
+                acquisition.state in {"starting", "awaiting_external_start"}
+                and acquisition.control_mode == "plugin-commanded"
+                and prepared_start_worker_outcome == "settled"
+            ):
+                # The terminal worker plus the dispatch gate proved that the
+                # durable START never crossed MQTT publish. Its checkpoint is
+                # already ceased above, so only local capture resources remain.
+                acquisition.transition(
+                    "failed",
+                    message_code="acquisition.start.physical_command_not_dispatched",
+                    result={
+                        "receiver_stopped": False,
+                        "device_state": "ready",
+                        "device_start": "not-dispatched",
+                        "automatic_command_retry": False,
+                    },
+                )
+                camera_terminal_status = "failed"
+                camera_failure_code = "physical-start-not-dispatched"
+                stop_runtime_for_camera_failure = True
+                zero_dispatch_start_operation_id = self._acquisition_start_operation_id
             elif (
                 acquisition.state in {"starting", "awaiting_external_start"}
                 and acquisition.control_mode == "plugin-commanded"
@@ -28860,6 +29030,20 @@ class XgridsK1CompatibilityService:
                             )
                             acquisition.state_revision += 1
                             acquisition.updated_at = datetime.now(UTC)
+                elif zero_dispatch_start_operation_id is not None:
+                    with self._lock:
+                        if (
+                            acquisition.state == "failed"
+                            and isinstance(acquisition.result, dict)
+                            and acquisition.result.get("device_start")
+                            == "not-dispatched"
+                        ):
+                            acquisition.result = {
+                                **acquisition.result,
+                                "receiver_stopped": True,
+                            }
+                            acquisition.state_revision += 1
+                            acquisition.updated_at = datetime.now(UTC)
                 elif confirmed_stop_local_camera_failure:
                     with self._lock:
                         if acquisition.state not in TERMINAL_ACQUISITION_STATES:
@@ -29032,6 +29216,35 @@ class XgridsK1CompatibilityService:
                     "automatic_replay_allowed": False,
                 },
             )
+        if zero_dispatch_start_operation_id is not None:
+            cleanup_succeeded = reconciliation_error is None
+            self._operations.transition_if_pending(
+                zero_dispatch_start_operation_id,
+                "failed",
+                stage_code=(
+                    "physical-start-not-dispatched"
+                    if cleanup_succeeded
+                    else "physical-start-not-dispatched-local-cleanup-failed"
+                ),
+                message_code=(
+                    "acquisition.start.physical_command_not_dispatched"
+                    if cleanup_succeeded
+                    else "acquisition.start.local_cleanup_failed"
+                ),
+                error={
+                    "category": "device" if cleanup_succeeded else "stream",
+                    "code": (
+                        "physical-start-not-dispatched"
+                        if cleanup_succeeded
+                        else "local-cleanup-failed-after-start-not-dispatched"
+                    ),
+                    "retryable": cleanup_succeeded,
+                    "safe_to_retry": cleanup_succeeded,
+                    "side_effect_status": "none",
+                    "physical_command_sent": False,
+                    "automatic_replay_allowed": False,
+                },
+            )
         if failed_operation_id is not None:
             self._operations.transition_if_pending(
                 failed_operation_id,
@@ -29169,6 +29382,11 @@ class XgridsK1CompatibilityService:
                 failed_operation_id,
                 unknown_start_operation_id,
                 recovery_standby_start_operation_id,
+                (
+                    zero_dispatch_start_operation_id
+                    if reconciliation_error is None
+                    else None
+                ),
             }:
                 self._acquisition_start_operation_id = None
             if self._acquisition_stop_operation_id == failed_stop_operation_id:

tests/test_xgrids_acquisition_lifecycle.py

@@ -90,6 +90,7 @@ from k1link.device_plugins.xgrids_k1.network_provisioning_idempotency_journal im
 )
 from k1link.device_plugins.xgrids_k1.physical_command_coordinator import (
     LedgerPhysicalCommandCoordinator,
+    PhysicalCommandIntentContext,
     PhysicalCommandRuntimeBinding,
 )
 from k1link.device_plugins.xgrids_k1.physical_command_ledger import (
@@ -7502,6 +7503,151 @@ def test_plugin_commanded_acquisition_keeps_start_and_stop_as_explicit_actions(
     )
 
 
+def test_terminal_prepared_start_without_publish_settles_and_stops_local_capture(
+    tmp_path: Path,
+) -> None:
+    service, runtime = service_with_fake_runtime(tmp_path)
+    control = FakeInteractiveControlSession()
+    service._application_control_session = control  # type: ignore[assignment]  # noqa: SLF001
+    binding = _seed_supervised_connection(
+        service,
+        transport_ref="k1-start-zero-publish",
+    )
+    prepared = service.prepare_acquisition(
+        _prepare_request(
+            project_name="START_NONE",
+            host=binding.target_ipv4,
+            compatibility_attestation=ATTESTATION,
+        )
+    )
+    acquisition_id = str(prepared["acquisition"]["acquisition_id"])
+    identity = PhysicalCommandIdentity(
+        vendor_device_id_sha256="a" * 64,
+        device_serial_sha256="b" * 64,
+    )
+    assert control.verified_control is not None
+    connection = PhysicalCommandConnectionBinding(
+        intent_id=binding.intent_id,
+        transport_ref=binding.transport_ref,
+        connection_mode=binding.connection_mode,
+        target_ipv4=binding.target_ipv4,
+        target_port=binding.target_port,
+        host_path_epoch=binding.host_path_epoch,
+        control_session_id=str(control.verified_control["control_session_id"]),
+        producer_generation=int(control.verified_control["producer_generation"]),
+    )
+    coordinator = service._physical_command_coordinator  # noqa: SLF001
+    coordinator.application_response(
+        ApplicationMqttResponseEvidence(
+            operation_key="bootstrap:start-none:DeviceInfoRequest",
+            response_topic="lixel/application/response/device_info",
+            payload_sha256="c" * 64,
+            modeling_action=None,
+            result_code=None,
+            success=None,
+            observed_at_utc="2026-08-14T13:30:40.000Z",
+        )
+    )
+    coordinator.bind_control_session(
+        PhysicalCommandRuntimeBinding(
+            vendor_device_id_sha256=identity.vendor_device_id_sha256,
+            device_serial_sha256=identity.device_serial_sha256,
+            compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
+            intent_id=connection.intent_id,
+            transport_ref=connection.transport_ref,
+            connection_mode=connection.connection_mode,
+            target_ipv4=connection.target_ipv4,
+            target_port=connection.target_port,
+            host_path_epoch=connection.host_path_epoch,
+            control_session_id=connection.control_session_id,
+            producer_generation=connection.producer_generation,
+        )
+    )
+    coordinator.device_status(
+        ApplicationMqttDeviceStatusEvidence(
+            vendor_device_id_sha256=identity.vendor_device_id_sha256,
+            device_serial_sha256=identity.device_serial_sha256,
+            session_state="ready",
+            session_state_code=MODELING_STATE_BASE + 300,
+            project_bound=False,
+            project_id_sha256=None,
+            init_ready=False,
+            status_message_sha256="d" * 64,
+            mqtt_retained=False,
+            observed_at_utc="2026-08-14T13:30:41.000Z",
+        )
+    )
+    payload = b"exact-start-zero-publish"
+    envelope = OneShotPublishEnvelope(
+        operation_key="modeling:start",
+        topic="lixel/application/request/modeling",
+        payload=payload,
+        payload_sha256=hashlib.sha256(payload).hexdigest(),
+        payload_bytes=len(payload),
+        qos=2,
+        retain=False,
+    )
+
+    def request_start_after_durable_prepare(**kwargs: object) -> dict[str, object]:
+        context = kwargs["command_context"]
+        observer = kwargs["preparation_checkpoint_observer"]
+        assert isinstance(context, PhysicalCommandIntentContext)
+        assert callable(observer)
+        coordinator.prepare(context, action="start", envelope=envelope)
+        observer("prepared", context, envelope)
+        control._accept_checkpoint(  # noqa: SLF001
+            expected_session_generation=kwargs["expected_session_generation"],  # type: ignore[arg-type]
+            expected_state_revision=kwargs["expected_state_revision"],  # type: ignore[arg-type]
+        )
+        control.start_projects.append(str(kwargs["project_name"]))
+        control.start_contexts.append(context)
+        control.state = "start-requested"
+        return control.snapshot()
+
+    control.request_start = request_start_after_durable_prepare  # type: ignore[method-assign]
+    starting = service.start_acquisition(
+        _start_request(
+            acquisition_id=acquisition_id,
+            physical_acceptance=PHYSICAL_ACCEPTANCE,
+        )
+    )
+    start_operation_id = str(starting["last_operation"]["operation_id"])
+    assert starting["physical_command"]["record"]["stage"] == "prepared"
+    assert starting["active_acquisition_recovery_checkpoint"]["state"] == "prepared"
+    assert runtime.source_mode == "live"
+
+    control.state = "failed"
+    control.state_revision += 1
+    control.failure = {
+        "reason_code": "application-connection-binding-lost",
+        "failed_phase": "start-requested",
+        "modeling_command_attempted": False,
+        "diagnostic_evidence_unavailable": [],
+        "safe_to_retry": True,
+    }
+    control.outcome_unknown = False
+    settled = service.state()
+
+    record = service._physical_command_ledger.snapshot().record  # noqa: SLF001
+    checkpoint = service._active_acquisition_checkpoint.snapshot().checkpoint  # type: ignore[union-attr]  # noqa: SLF001
+    operation = service._operations.get(start_operation_id)  # noqa: SLF001
+    assert record is not None
+    assert record.stage == "resolved"
+    assert record.resolution == "not-dispatched"
+    assert checkpoint is not None and checkpoint.state == "ceased"
+    assert settled["acquisition"]["state"] == "failed"
+    assert settled["acquisition"]["result"]["device_start"] == "not-dispatched"
+    assert settled["acquisition"]["result"]["receiver_stopped"] is True
+    assert operation.status == "failed"
+    assert operation.stage_code == "physical-start-not-dispatched"
+    assert operation.error is not None
+    assert operation.error["side_effect_status"] == "none"
+    assert operation.error["physical_command_sent"] is False
+    assert runtime.stop_calls == 1
+    assert runtime.source_mode == "idle"
+    assert service._acquisition_session_lease is None  # noqa: SLF001
+
+
 def _activate_real_checkpoint_for_prepared_stop_fixture(
     service: XgridsK1CompatibilityService,
     *,
@@ -12879,6 +13025,11 @@ def test_public_commanded_workflow_refreshes_stable_route_before_ttl_reduction(
     control, binding, ready = _install_binding_validating_ready_control(service)
     stable_path = _direct_host_path(binding.target_ipv4)
     monkeypatch.setattr(service, "_sample_host_path", lambda *_args, **_kwargs: stable_path)
+    monkeypatch.setattr(
+        facade_module,
+        "_probe_control_endpoint_socket",
+        lambda _target: facade_module.TcpReachabilityProbeResult(reachable=True),
+    )
     target = EndpointTarget(binding.target_ipv4, binding.target_port)
 
     def advance_past_fifteen_seconds_with_fresh_endpoint() -> None:
@@ -13244,6 +13395,82 @@ def test_start_refreshes_ttl_expired_endpoint_after_project_prepare(
     assert control.start_projects == [PROJECT_NAME]
 
 
+def test_start_refreshes_endpoint_before_prepared_worker_handoff_window(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    service, runtime = service_with_fake_runtime(tmp_path)
+    monotonic_now = [100.0]
+    suspend_aware_now = [1_000.0]
+    supervisor = service._connection_supervisor  # noqa: SLF001
+    supervisor._monotonic_clock = lambda: monotonic_now[0]  # noqa: SLF001
+    supervisor._suspend_aware_clock = lambda: suspend_aware_now[0]  # noqa: SLF001
+    supervisor._observation_ttl_seconds = 15.0  # noqa: SLF001
+    control, binding, ready = _install_binding_validating_ready_control(service)
+    stable_path = _direct_host_path(binding.target_ipv4)
+    monkeypatch.setattr(service, "_sample_host_path", lambda *_args, **_kwargs: stable_path)
+    workspace = service.enter_application_workspace(
+        EnterApplicationWorkspaceRequest(
+            operator_confirmed=True,
+            expected_session_generation=ready["application_control_session"][
+                "session_generation"
+            ],
+            expected_state_revision=ready["application_control_session"]["state_revision"],
+        )
+    )
+    prepared = service.prepare_acquisition(
+        _prepare_request(
+            project_name=PROJECT_NAME,
+            host=binding.target_ipv4,
+            compatibility_attestation=ATTESTATION,
+            expected_control_session_generation=workspace["application_control_session"][
+                "session_generation"
+            ],
+            expected_control_state_revision=workspace["application_control_session"][
+                "state_revision"
+            ],
+        )
+    )
+    # The endpoint is still nominally reachable, but only eleven seconds of
+    # its lease remain. That is insufficient for PREPARED + worker dispatch.
+    monotonic_now[0] += 4.01
+    suspend_aware_now[0] += 4.01
+    assert supervisor.snapshot().endpoint.tcp_state == "reachable"
+    tcp_samples: list[str] = []
+    monkeypatch.setattr(
+        facade_module,
+        "_probe_control_endpoint_socket",
+        lambda target: (
+            tcp_samples.append(target)
+            or facade_module.TcpReachabilityProbeResult(reachable=True)
+        ),
+    )
+
+    started = service.start_acquisition(
+        _start_request(
+            acquisition_id=prepared["acquisition"]["acquisition_id"],
+            physical_acceptance=PHYSICAL_ACCEPTANCE,
+            expected_control_session_generation=prepared["application_control_session"][
+                "session_generation"
+            ],
+            expected_control_state_revision=prepared["application_control_session"][
+                "state_revision"
+            ],
+        )
+    )
+
+    assert started["acquisition"]["state"] == "starting"
+    assert tcp_samples == [binding.target_ipv4]
+    assert supervisor.endpoint_observation_has_remaining_lease(
+        target=EndpointTarget(binding.target_ipv4, binding.target_port),
+        intent_id=binding.intent_id,
+        host_path_epoch=binding.host_path_epoch,
+        minimum_remaining_seconds=facade_module.COMMAND_ENDPOINT_MIN_REMAINING_SECONDS,
+    )
+    assert len(runtime.start_calls) == 1
+    assert control.start_projects == [PROJECT_NAME]
+
+
 def test_start_stale_control_proof_fails_before_receiver_or_device_checkpoint(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,