diff --git a/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py b/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py index 0aeab42..f5e21f8 100644 --- a/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py +++ b/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, *, diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index 5346674..ba69ae4 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/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 - # TCP silence lease while the exact MQTT control proof remains - # fresh. Refresh only that expired read-only endpoint fact before - # the physical START admission. This is not a command retry: no - # operation or physical ledger row exists yet. A second route - # sample binds the TCP result to the same host/association epoch; - # any changed path or unreachable broker still fails closed. + # Project preparation can outlive the supervisor's TCP silence + # lease, or leave too little lease for durable PREPARED plus the + # worker publish guard. Refresh only that exact read-only endpoint + # fact. This is not a command retry. A second route sample binds + # the TCP result to the same host/association epoch; 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.acquisition_id == context.acquisition_id + and record.operation_id == operation_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.original_start_operation_id == context.operation_id - and checkpoint.start_payload_sha256 == envelope.payload_sha256 + and checkpoint.acquisition_id == acquisition_id + and checkpoint.original_start_operation_id == operation_id + 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: diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index 02f0535..d4b1c42 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/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,