diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cc30dee8..8a1a8158 100644 --- a/.github/workflows/ci.yml +++ b/.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: | diff --git a/services/process_control/app/protocols/secs_gem/transport.py b/services/process_control/app/protocols/secs_gem/transport.py index 0e8f726c..10f54609 100644 --- a/services/process_control/app/protocols/secs_gem/transport.py +++ b/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 diff --git a/services/process_control/app/simulators/ion_implant_hil.py b/services/process_control/app/simulators/ion_implant_hil.py index ba361ae5..136694f9 100644 --- a/services/process_control/app/simulators/ion_implant_hil.py +++ b/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 ) diff --git a/services/process_control/app/simulators/rtp_hil.py b/services/process_control/app/simulators/rtp_hil.py index 4e655d66..d65ca035 100644 --- a/services/process_control/app/simulators/rtp_hil.py +++ b/services/process_control/app/simulators/rtp_hil.py @@ -1,5 +1,7 @@ """RTP Hardware-in-Loop (HIL) Simulator with thermal plant model.""" +import math + import numpy as np from typing import Dict, List, Optional, Any from dataclasses import dataclass @@ -44,6 +46,10 @@ class ThermalZoneState: temperature_C: float lamp_power_pct: float time_constant_s: float + # Power the lamps are actually delivering — first-order response toward + # the commanded power with this zone's time constant (tungsten filaments + # do not step instantaneously). + applied_power_pct: float = 0.0 class ThermalPlantModel: @@ -89,9 +95,19 @@ def __init__(self, num_zones: int = 4, random_seed: Optional[int] = None): self.gas_flow_sccm = 0.0 self.chamber_pressure_torr = 760.0 - # Controller state + # Controller state. setpoint_C is the PROFILED setpoint the PID + # tracks each step; target_C is where the profile is headed, at + # setpoint_ramp_rate_C_per_s (None = step change). This is what + # honors set_target_temperature(..., ramp_rate_C_per_s=...) — the + # old model silently ignored the requested ramp rate. self.setpoint_C = 25.0 - self.max_lamp_power_W = 10000.0 + self.target_C = 25.0 + self.setpoint_ramp_rate_C_per_s: Optional[float] = None + # 10 kW per zone (realistic tungsten-halogen bank for 300 mm RTP). + # Steady state at 1000C needs ~11 kW total against radiative + + # convective losses; at the old 10 kW TOTAL the plant equilibrated + # near 450C with lamps saturated and could never reach setpoint. + self.max_lamp_power_W = 10000.0 * num_zones self.lamp_saturation_reached = False # PID controller (for automatic control) @@ -114,9 +130,28 @@ def update(self, dt: float): """ self.simulation_time_s += dt + # Advance the profiled setpoint toward the target + if self.setpoint_C != self.target_C: + rate = self.setpoint_ramp_rate_C_per_s + if rate is None or rate <= 0: + self.setpoint_C = self.target_C + else: + step = rate * dt + delta = self.target_C - self.setpoint_C + self.setpoint_C = ( + self.target_C + if abs(delta) <= step + else self.setpoint_C + math.copysign(step, delta) + ) + # Calculate target temperature based on setpoint and PID error = self.setpoint_C - self.wafer_temp_C self.integral_error += error * dt + # Anti-windup: bound the I-term to actuator authority (+/-100% at + # ki=0.1 -> |integral| <= 1000). Without this, a long saturated ramp + # wound up thousands of %-seconds and the hold oscillated for hours. + integral_limit = 100.0 / self.ki if self.ki > 0 else 0.0 + self.integral_error = max(-integral_limit, min(integral_limit, self.integral_error)) derivative_error = (error - self.last_error) / dt if dt > 0 else 0 self.last_error = error @@ -130,31 +165,45 @@ def update(self, dt: float): zone_temps = [zone.temperature_C for zone in self.zones] for i, zone in enumerate(self.zones): - # Apply lamp power with zone-specific response + # Commanded power: manual zone override, else the PID demand target_power = zone.lamp_power_pct if zone.lamp_power_pct > 0 else base_power_pct # Check for actuator saturation self.lamp_saturation_reached = target_power >= 99.0 + # Lamp response: first-order approach toward the command with + # this zone's time constant (the lag belongs on the actuator, + # not on the temperature state). + zone.applied_power_pct += (target_power - zone.applied_power_pct) * min( + 1.0, dt / zone.time_constant_s + ) + # Power to heat (simplified model) - power_W = (target_power / 100.0) * (self.max_lamp_power_W / self.num_zones) + power_W = (zone.applied_power_pct / 100.0) * (self.max_lamp_power_W / self.num_zones) # Heat input heat_input_J_per_s = power_W + # Each zone covers 1/num_zones of the wafer — power, heat + # capacity, AND loss area must all use the same share. The old + # code lost heat over the FULL wafer area in every zone (a 4x + # double-count) while heating with 1/4 of the lamps, so the + # plant saturated ~40C below a 1000C setpoint forever. + zone_area_m2 = WAFER_AREA_M2 / self.num_zones + # Radiative cooling (Stefan-Boltzmann) T_kelvin = zone.temperature_C + 273.15 T_ambient_kelvin = 298.15 # 25°C radiative_loss = ( self.emissivity * STEFAN_BOLTZMANN - * WAFER_AREA_M2 + * zone_area_m2 * (T_kelvin**4 - T_ambient_kelvin**4) ) # Convective cooling (depends on gas flow and pressure) convective_coeff = self._calculate_convective_coefficient() - convective_loss = convective_coeff * WAFER_AREA_M2 * (zone.temperature_C - 25.0) + convective_loss = convective_coeff * zone_area_m2 * (zone.temperature_C - 25.0) # Net heat flux net_heat_flux = heat_input_J_per_s - radiative_loss - convective_loss @@ -163,11 +212,10 @@ def update(self, dt: float): wafer_heat_capacity = WAFER_MASS_KG * SI_SPECIFIC_HEAT / self.num_zones temp_rate = net_heat_flux / wafer_heat_capacity - # Update zone temperature with time constant - zone_target_temp = zone.temperature_C + temp_rate * dt - zone.temperature_C += (zone_target_temp - zone.temperature_C) * ( - dt / zone.time_constant_s - ) + # Direct Euler integration — the wafer's heat capacity IS the + # thermal inertia; adding another first-order lag here made the + # response depend on dt (dt^2/tau) and physically ~25x too slow. + zone.temperature_C += temp_rate * dt # Thermal coupling between zones (heat diffusion) if i > 0: @@ -269,9 +317,13 @@ def set_lamp_power(self, zone_powers: List[float]): for zone, power in zip(self.zones, zone_powers): zone.lamp_power_pct = max(0, min(100, power)) - def set_setpoint(self, setpoint_C: float): - """Set temperature setpoint.""" - self.setpoint_C = setpoint_C + def set_setpoint(self, setpoint_C: float, ramp_rate_C_per_s: Optional[float] = None): + """Set temperature target; ramp the profiled setpoint at the given + rate (None = step change, preserving legacy callers).""" + self.target_C = setpoint_C + self.setpoint_ramp_rate_C_per_s = ramp_rate_C_per_s + if ramp_rate_C_per_s is None or ramp_rate_C_per_s <= 0: + self.setpoint_C = setpoint_C def get_overshoot(self) -> float: """Calculate temperature overshoot percentage.""" @@ -288,9 +340,15 @@ def reset(self): zone.temperature_C = 25.0 zone.lamp_power_pct = 0.0 + for zone in self.zones: + zone.applied_power_pct = 0.0 + self.wafer_temp_C = 25.0 self.pyrometer_temp_C = 25.0 self.thermocouple_temp_C = 25.0 + self.setpoint_C = 25.0 + self.target_C = 25.0 + self.setpoint_ramp_rate_C_per_s = None self.integral_error = 0.0 self.last_error = 0.0 self.wafer_temp_history = [25.0] @@ -310,10 +368,16 @@ def __init__( num_zones: int = 4, random_seed: Optional[int] = None, simulation_timestep_s: float = 0.1, + time_acceleration: float = 1.0, ): super().__init__(equipment_id) self.num_zones = num_zones self.simulation_timestep_s = simulation_timestep_s + # Virtual clock (Phase 4.4): simulated time advances at + # time_acceleration x wall time. 1.0 = real time (production); + # soak tests run 1000x so "12 hours" of physics happens in 43s. + self.time_acceleration = time_acceleration + self._sim_residual_s = 0.0 # Thermal plant self.thermal_plant = ThermalPlantModel(num_zones=num_zones, random_seed=random_seed) @@ -330,6 +394,11 @@ def __init__( self._recipe_start_time: Optional[datetime] = None self._current_segment_index = 0 self._segment_start_time: Optional[datetime] = None + # Final progress of the last completed run. Natural completion + # calls stop_recipe(), which clears _run_id — without this, a + # finished recipe reported progress_pct 0.0 (indistinguishable + # from never having run) and the soak tests scored 0% completion. + self._last_run_summary: Optional[Dict[str, Any]] = None # Last update time self._last_sim_update = datetime.now() @@ -362,17 +431,28 @@ async def shutdown(self) -> bool: self.status = RTPStatus.SHUTDOWN return True + # One update() call may only do this many plant steps; anything beyond + # stays in the residual for the next call. Bounds a single call's work + # without ever LOSING simulated time (the old code truncated + # int(elapsed/dt) and reset the clock, so any poll faster than one + # timestep silently discarded time — sim progress depended on caller + # cadence, which is what made the soak tests flake under load). + MAX_STEPS_PER_UPDATE = 200_000 + def _update_simulation(self): - """Update thermal simulation based on elapsed time.""" + """Advance the thermal simulation by accumulated (virtual) time.""" now = datetime.now() elapsed = (now - self._last_sim_update).total_seconds() + self._last_sim_update = now - # Run multiple simulation steps if needed - num_steps = int(elapsed / self.simulation_timestep_s) - for _ in range(min(num_steps, 100)): # Limit to prevent runaway + self._sim_residual_s += elapsed * self.time_acceleration + num_steps = min( + int(self._sim_residual_s / self.simulation_timestep_s), + self.MAX_STEPS_PER_UPDATE, + ) + for _ in range(num_steps): self.thermal_plant.update(self.simulation_timestep_s) - - self._last_sim_update = now + self._sim_residual_s -= num_steps * self.simulation_timestep_s # Sync helpers for the execute task ------------------------------------ # (Phase 1.2: execute_rtp_run always called get_telemetry()/ @@ -404,16 +484,20 @@ async def set_target_temperature( self, temp_C: float, ramp_rate_C_per_s: Optional[float] = None ) -> bool: logger.info(f"HIL: Setting target temperature: {temp_C}°C") - self.thermal_plant.set_setpoint(temp_C) + self.thermal_plant.set_setpoint(temp_C, ramp_rate_C_per_s=ramp_rate_C_per_s) - # Update status - current_temp = self.thermal_plant.wafer_temp_C - if temp_C > current_temp + 5: - self.status = RTPStatus.HEATING - elif temp_C < current_temp - 5: - self.status = RTPStatus.COOLING - else: - self.status = RTPStatus.AT_TEMPERATURE + # Update status — but never clobber an active recipe run: + # start_recipe()/segment advances call this for every segment, and + # overwriting RUNNING_RECIPE with HEATING made get_recipe_progress + # report is_running=False from the very first poll. + if self.status != RTPStatus.RUNNING_RECIPE: + current_temp = self.thermal_plant.wafer_temp_C + if temp_C > current_temp + 5: + self.status = RTPStatus.HEATING + elif temp_C < current_temp - 5: + self.status = RTPStatus.COOLING + else: + self.status = RTPStatus.AT_TEMPERATURE return True @@ -516,6 +600,7 @@ async def start_recipe(self, recipe_id: str) -> str: self._recipe_start_time = datetime.now() self._segment_start_time = datetime.now() self._current_segment_index = 0 + self._last_run_summary = None self.status = RTPStatus.RUNNING_RECIPE # Start first segment @@ -548,6 +633,8 @@ async def get_recipe_progress(self) -> Dict[str, Any]: self._update_simulation() if self._current_recipe is None or self._run_id is None: + if self._last_run_summary is not None: + return dict(self._last_run_summary) return { "is_running": False, "run_id": None, @@ -559,7 +646,7 @@ async def get_recipe_progress(self) -> Dict[str, Any]: } elapsed = ( - (datetime.now() - self._recipe_start_time).total_seconds() + (datetime.now() - self._recipe_start_time).total_seconds() * self.time_acceleration if self._recipe_start_time else 0.0 ) @@ -568,7 +655,9 @@ async def get_recipe_progress(self) -> Dict[str, Any]: # Check if we should advance to next segment if self._segment_start_time and self._current_segment_index < num_segments: segment = self._current_recipe.segments[self._current_segment_index] - segment_elapsed = (datetime.now() - self._segment_start_time).total_seconds() + segment_elapsed = ( + datetime.now() - self._segment_start_time + ).total_seconds() * self.time_acceleration # Estimate time to reach target current_temp = self.thermal_plant.wafer_temp_C @@ -590,7 +679,20 @@ async def get_recipe_progress(self) -> Dict[str, Any]: next_segment.target_temp_C, next_segment.ramp_rate_C_per_s ) else: - # Recipe complete + # Recipe complete — record the final summary before + # stop_recipe() clears the run id. + self._last_run_summary = { + "is_running": False, + "run_id": self._run_id, + "recipe_name": self._current_recipe.recipe_name, + "current_segment": self._current_segment_index, + "total_segments": num_segments, + "elapsed_time_s": elapsed, + "progress_pct": 100.0, + "current_temp_C": self.thermal_plant.wafer_temp_C, + "target_temp_C": self.thermal_plant.setpoint_C, + "overshoot_pct": self.thermal_plant.get_overshoot(), + } await self.stop_recipe() return { diff --git a/services/process_control/tests/soak_tests/test_ion_implant_soak.py b/services/process_control/tests/soak_tests/test_ion_implant_soak.py index 37e65d03..efde0a85 100644 --- a/services/process_control/tests/soak_tests/test_ion_implant_soak.py +++ b/services/process_control/tests/soak_tests/test_ion_implant_soak.py @@ -38,7 +38,12 @@ class SoakTestConfig: async def ion_implant_system(): """Create ion implant HIL system for testing.""" driver = IonImplantHILDriver( - equipment_id="SOAK-TEST-ION-01", random_seed=42, wafer_diameter_mm=300.0 # Deterministic + equipment_id="SOAK-TEST-ION-01", + random_seed=42, # Deterministic + wafer_diameter_mm=300.0, + # Virtual clock: dose integrates over simulated time, so per-wafer + # budgets like "60 simulated seconds" are actually 60ms of wall time. + time_acceleration=SoakTestConfig.TIME_ACCELERATION, ) telemetry_manager = IonImplantTelemetryManager( @@ -181,10 +186,6 @@ async def test_ion_implant_12h_stability(ion_implant_system): @pytest.mark.asyncio @pytest.mark.soak @pytest.mark.timeout(120) # Real-time timeout for 24h accelerated test -@pytest.mark.xfail( - reason="Phase 4.4: timing-coupled accelerated-time sim — outcome varies with host scheduling (observed both pass and fail across identical clean runs at seed 42); needs a virtual-clock rework to be deterministic", - strict=False, -) async def test_ion_implant_24h_multiple_wafers(ion_implant_system): """ 24-hour soak test: Multiple wafer processing. @@ -323,10 +324,6 @@ async def test_ion_implant_24h_multiple_wafers(ion_implant_system): @pytest.mark.soak @pytest.mark.slow @pytest.mark.timeout(300) # Real-time timeout for 72h accelerated test -@pytest.mark.xfail( - reason="Phase 4.4: timing-coupled accelerated-time sim — outcome varies with host scheduling (observed both pass and fail across identical clean runs at seed 42); needs a virtual-clock rework to be deterministic", - strict=False, -) async def test_ion_implant_72h_stress(ion_implant_system): """ 72-hour stress test: Extreme conditions and recovery. diff --git a/services/process_control/tests/soak_tests/test_rtp_soak.py b/services/process_control/tests/soak_tests/test_rtp_soak.py index ac7b6fda..7e72eb32 100644 --- a/services/process_control/tests/soak_tests/test_rtp_soak.py +++ b/services/process_control/tests/soak_tests/test_rtp_soak.py @@ -41,6 +41,9 @@ async def rtp_system(): num_zones=4, random_seed=42, # Deterministic simulation_timestep_s=0.1, + # Virtual clock: the sim itself runs accelerated, so the "12h" + # asserts below measure 12 simulated hours, not 43 wall seconds. + time_acceleration=SoakTestConfig.TIME_ACCELERATION, ) telemetry_manager = RTPTelemetryManager( @@ -59,10 +62,6 @@ async def rtp_system(): @pytest.mark.asyncio @pytest.mark.soak @pytest.mark.timeout(60) # Real-time timeout for 12h accelerated test -@pytest.mark.xfail( - reason="Phase 4.4: RTP thermal model spec-vs-implementation disagreement (pyrometer std ~45C vs <5C spec) — same family as the test_rtp_thermal.py/test_rtp_controllers.py CI exclusions", - strict=False, -) async def test_rtp_12h_thermal_stability(rtp_system): """ 12-hour soak test: Continuous temperature hold. @@ -183,10 +182,6 @@ async def test_rtp_12h_thermal_stability(rtp_system): @pytest.mark.asyncio @pytest.mark.soak @pytest.mark.timeout(120) # Real-time timeout for 24h accelerated test -@pytest.mark.xfail( - reason="Phase 4.4: RTP thermal model spec-vs-implementation disagreement (pyrometer std ~45C vs <5C spec) — same family as the test_rtp_thermal.py/test_rtp_controllers.py CI exclusions", - strict=False, -) async def test_rtp_24h_thermal_cycling(rtp_system): """ 24-hour soak test: Repeated thermal cycles. @@ -321,10 +316,6 @@ async def test_rtp_24h_thermal_cycling(rtp_system): @pytest.mark.soak @pytest.mark.slow @pytest.mark.timeout(300) # Real-time timeout for 72h accelerated test -@pytest.mark.xfail( - reason="Phase 4.4: RTP thermal model spec-vs-implementation disagreement (pyrometer std ~45C vs <5C spec) — same family as the test_rtp_thermal.py/test_rtp_controllers.py CI exclusions", - strict=False, -) async def test_rtp_72h_recipe_stress(rtp_system): """ 72-hour stress test: Complex recipes under various conditions. @@ -408,9 +399,19 @@ async def test_rtp_72h_recipe_stress(rtp_system): if not progress["is_running"]: break - # Timeout check (2x expected time) - expected_time = sum(seg.dwell_time_s for seg in recipe_spec["segments"]) - timeout = expected_time * 2 / SoakTestConfig.TIME_ACCELERATION + # Timeout: 2x the full expected duration — dwells AND ramps + # (the old math counted only dwells, so a recipe whose ramps + # dominate always "timed out"), with a 300-sim-second floor + # because the final cool-to-ambient approaches asymptotically. + prev_temp = 25.0 + expected_time = 0.0 + for seg in recipe_spec["segments"]: + expected_time += ( + abs(seg.target_temp_C - prev_temp) / max(seg.ramp_rate_C_per_s, 0.1) + + seg.dwell_time_s + ) + prev_temp = seg.target_temp_C + timeout = max(expected_time * 2, 300.0) / SoakTestConfig.TIME_ACCELERATION if (datetime.now() - recipe_start).total_seconds() > timeout: print( @@ -419,7 +420,9 @@ async def test_rtp_72h_recipe_stress(rtp_system): await driver.stop_recipe() break - await asyncio.sleep(0.5) + # Poll fine-grained: at 1000x acceleration a 0.5s real sleep is + # 500 simulated seconds — coarser than the whole recipe. + await asyncio.sleep(0.02) # Record result recipe_result = { diff --git a/services/process_control/tests/unit/test_equipment_manager.py b/services/process_control/tests/unit/test_equipment_manager.py index 70853f85..bc1481e8 100644 --- a/services/process_control/tests/unit/test_equipment_manager.py +++ b/services/process_control/tests/unit/test_equipment_manager.py @@ -486,11 +486,6 @@ async def test_equipment_handler_ignores_unsupported_streams( await server.stop() -@pytest.mark.skip( - reason="Async-timing flake: handler-exit await races the closed-connection " - "signal (CancelledError/TimeoutError under load). Tracked in Phase 4.4 — " - "needs a deterministic close-event instead of sleep-based settling." -) async def test_equipment_handler_exits_when_connection_closes( db_session: Session, session_factory,