Retain K1 connection between scans and admit the next named acquisition

tests/test_xgrids_application_session.py

@@ -372,6 +372,14 @@ class FakeExecutor:
     def maintain_post_stop_until_standby(self) -> None:
         self.records.append("wait:device-standby")
 
+    def maintain_standby_until_next_acquisition(self, requested: Callable[[], bool]) -> None:
+        while not requested():
+            if self.transport.state == "closed":
+                raise RuntimeError("test transport closed")
+            if not self.transport.proof_fresh:
+                raise ApplicationControlProofStale("test control proof expired")
+            threading.Event().wait(0.005)
+
     def snapshot(self) -> dict[str, object]:
         return {
             "records": list(self.records),
@@ -2160,9 +2168,11 @@ def test_new_explicit_session_waits_for_old_worker_transport_retirement(
     _wait_phase(session, "scanning")
     session.request_stop(confirmation=_confirmation())
     _wait_phase(session, "completed")
+    assert new_thread.is_alive()
+    session.close()
     new_thread.join(timeout=2.0)
     assert not new_thread.is_alive()
-    assert transports[1].close_calls == 1
+    assert transports[1].close_calls == 2  # explicit close and idempotent worker cleanup
 
 
 def test_start_outcome_unknown_blocks_reopen_even_when_transport_is_closed(
@@ -2856,3 +2866,66 @@ def test_pretransport_authority_failure_remains_safe_after_worker_retirement() -
     assert failure["diagnostic_snapshot_unavailable"] == []
     assert failure["diagnostic_evidence_unavailable"] == []
     assert failure["safe_to_retry"] is True
+
+
+@pytest.fixture(autouse=True)
+def close_test_control_sessions(monkeypatch: pytest.MonkeyPatch):
+    sessions = []
+    original = InteractiveApplicationControlSession.__init__
+
+    def tracked(self, *args, **kwargs):
+        original(self, *args, **kwargs)
+        sessions.append(self)
+
+    monkeypatch.setattr(InteractiveApplicationControlSession, "__init__", tracked)
+    yield
+    for session in sessions:
+        session.close()
+
+
+def test_two_named_scans_retain_idle_connection_and_require_new_dialogues(monkeypatch):
+    FakeExecutor.records = []
+    monkeypatch.setattr(session_module, "PhysicalAcceptanceDialogueExecutor", FakeExecutor)
+    transports = []
+    def create_transport(host):
+        transport = FakeTransport(host)
+        transports.append(transport)
+        return transport
+    session = InteractiveApplicationControlSession(
+        FakeAuthorityLoader(), transport_factory=create_transport,
+    )
+    session.open(host="192.168.1.20", timezone_name="UTC", connection_binding=_connection_binding())
+    _wait_phase(session, "connection-ready")
+    first_checkpoint = None
+    for project in ("FIRST", "SECOND"):
+        current = session.snapshot()
+        if first_checkpoint is not None:
+            with pytest.raises(ApplicationAcceptanceError):
+                session.enter_workspace(expected_session_generation=first_checkpoint[0],
+                                        expected_state_revision=first_checkpoint[1])
+        first_checkpoint = (current["session_generation"], current["state_revision"])
+        session.enter_workspace(expected_session_generation=current["session_generation"],
+                                expected_state_revision=current["state_revision"])
+        _wait_phase(session, "workspace-ready")
+        session.open_project_prompt()
+        _wait_phase(session, "project-ready")
+        session.request_start(project_name=project, confirmation=_confirmation())
+        _wait_phase(session, "scanning")
+        session.request_stop(confirmation=_confirmation())
+        finished = _wait_phase(session, "completed")
+        assert finished["control_socket_open"] is True
+        assert finished["can_enter_workspace"] is True
+        assert finished["verified_control"]["control_proof_fresh"] is True
+        before = list(FakeExecutor.records)
+        threading.Event().wait(0.025)
+        assert FakeExecutor.records == before  # no timer-driven new acquisition
+        assert transports[-1].state == "ready"
+        assert len(transports) == (1 if project == "FIRST" else 2)
+    assert len(transports) == 2
+    assert transports[0].state == "closed"
+    assert FakeExecutor.records.count("start:11-14") == 2
+    assert FakeExecutor.records.count("stop") == 2
+    transports[-1].proof_fresh = False
+    _wait_phase(session, "failed")
+    assert session.snapshot()["control_socket_open"] is False
+    assert len(transports) == 2  # no reconnect or physical command replay on real loss