fix(pc): Phase 4.4 — virtual-clock HIL sims, real physics fixes, HSMS close sentinel

.github/workflows/ci.yml

@@ -119,12 +119,14 @@ jobs:
 
       # Phase 4.1: the FULL pc unit tree runs now that its models use the
       # cross-dialect type shims (the raw postgresql.UUID imports were the
-      # only thing keeping all 437 tests dormant). Five files with known
-      # physics/controller spec-vs-implementation disagreements are excluded
-      # and tracked for Phase 4.4:
-      #   test_ion_controllers.py (23), test_ion_range.py (12),
-      #   test_rtp_thermal.py (5), test_rtp_controllers.py (2),
-      #   (equipment_manager's async-flaky handler-exit test is skip-marked in-file)
+      # only thing keeping all 437 tests dormant). Phase 4.4 closed the
+      # remaining holes: the physics/controller files all pass (the comment
+      # here used to claim they were excluded — no exclusion mechanism ever
+      # existed and they run green), and the equipment_manager handler-exit
+      # test was un-skipped after the HSMS transport gained a deterministic
+      # close sentinel. The accelerated-time soak tests stay out of CI by
+      # design (7.5 min, load-sensitive) — run them locally via
+      # `pytest services/process_control/tests/soak_tests`.
       - name: Run process_control unit tests
         working-directory: services/process_control
         run: |

services/process_control/app/protocols/secs_gem/transport.py

@@ -132,7 +132,8 @@ def __init__(
         self._selected_event: Optional[asyncio.Event] = None
 
         # Data-message queue for the application to poll via recv_data_message.
-        self._data_queue: asyncio.Queue[SecsMessage] = asyncio.Queue()
+        # None = close sentinel; see recv_data_message.
+        self._data_queue: asyncio.Queue[Optional[SecsMessage]] = asyncio.Queue()
 
         # Monotonically incrementing system_bytes for outbound requests.
         # 32-bit wrap is fine — we never have 2^32 in-flight messages.
@@ -282,10 +283,22 @@ async def recv_data_message(self) -> SecsMessage:
         (Linktest.req, Select.req, ...) are handled by the reader task
         and never appear here. Replies to outbound requests are
         routed to their pending futures, also not surfaced here.
+
+        Raises HsmsConnectionError promptly when the link goes down —
+        the reader task and close() push a None sentinel through the
+        queue, so a coroutine already blocked here wakes deterministically
+        instead of hanging until someone cancels it (the Phase 4.4
+        equipment_handler flake). Messages queued before the close are
+        still delivered first (the sentinel sits behind them).
         """
-        if self.state == HsmsState.NOT_CONNECTED:
+        if self.state == HsmsState.NOT_CONNECTED and self._data_queue.empty():
             raise HsmsConnectionError("cannot recv on closed connection")
-        return await self._data_queue.get()
+        item = await self._data_queue.get()
+        if item is None:
+            # Re-arm so every subsequent (or concurrent) getter also wakes.
+            self._data_queue.put_nowait(None)
+            raise HsmsConnectionError("connection closed")
+        return item
 
     async def close(self) -> None:
         """Close the connection. Sends Separate.req if possible, then
@@ -339,6 +352,9 @@ async def close(self) -> None:
             self._pending_select_rsp.set_exception(HsmsConnectionError("connection closed"))
             self._pending_select_rsp = None
 
+        # Wake any coroutine blocked in recv_data_message.
+        self._data_queue.put_nowait(None)
+
     # ------------------------------------------------------------------
     # Internals — handshake
     # ------------------------------------------------------------------
@@ -412,6 +428,9 @@ async def _read_loop(self) -> None:
             logger.exception("HSMS reader task crashed")
         finally:
             self.state = HsmsState.NOT_CONNECTED
+            # Wake anyone blocked in recv_data_message — EOF/error exits
+            # previously left them hanging forever.
+            self._data_queue.put_nowait(None)
 
     async def _read_one_frame(self) -> Optional[bytes]:
         """Read one length-prefixed frame from the socket. Returns the

services/process_control/app/simulators/ion_implant_hil.py

@@ -338,11 +338,18 @@ class IonImplantHILDriver(IonImplantDriver):
     """Hardware-in-Loop driver with physics simulation."""
 
     def __init__(
-        self, equipment_id: str, random_seed: Optional[int] = None, wafer_diameter_mm: float = 300.0
+        self,
+        equipment_id: str,
+        random_seed: Optional[int] = None,
+        wafer_diameter_mm: float = 300.0,
+        time_acceleration: float = 1.0,
     ):
         super().__init__(equipment_id)
         self.physics = SRIMPhysicsModel(random_seed=random_seed)
         self.wafer_diameter_mm = wafer_diameter_mm
+        # Virtual clock (Phase 4.4): dose/beam time advances at
+        # time_acceleration x wall time. 1.0 = real time (production).
+        self.time_acceleration = time_acceleration
 
         # State variables
         self._source_params: Optional[SourceParameters] = None
@@ -481,7 +488,7 @@ async def set_beam_steering(self, x_offset_mm: float, y_offset_mm: float) -> boo
     async def get_beam_position(self) -> Tuple[float, float]:
         # Add jitter to position
         if self._last_update_time:
-            dt = (datetime.now() - self._last_update_time).total_seconds()
+            dt = (datetime.now() - self._last_update_time).total_seconds() * self.time_acceleration
             jittered_pos = self.physics.simulate_beam_jitter(self._beam_steering, dt)
             self._last_update_time = datetime.now()
             return jittered_pos
@@ -641,7 +648,9 @@ async def get_dose_integrator_reading(self) -> Dict:
             }
 
         if self.status == ImplantStatus.RUNNING and self._implant_start_time:
-            elapsed = (datetime.now() - self._implant_start_time).total_seconds()
+            elapsed = (
+                datetime.now() - self._implant_start_time
+            ).total_seconds() * self.time_acceleration
 
             # Simulate dose accumulation with noise
             ideal_dose_rate = (
@@ -668,7 +677,7 @@ async def get_dose_integrator_reading(self) -> Dict:
         )
         integrated_charge = self._current_dose * self._dose_params.wafer_area_cm2 * 1.6e-19
         elapsed = (
-            (datetime.now() - self._implant_start_time).total_seconds()
+            (datetime.now() - self._implant_start_time).total_seconds() * self.time_acceleration
             if self._implant_start_time
             else 0.0
         )