From b480b23d7770ec6922814b6ab9343f565e84301a Mon Sep 17 00:00:00 2001 From: braceal Date: Tue, 11 Aug 2026 16:34:44 -0500 Subject: [PATCH] Fix inverted use_stale_model gating in DDWEStreamThinker The two branches guarding inference submission were swapped relative to what the flag means. With use_stale_model=False the thinker only waited for train_output to become non-None, so it would happily submit inference against an arbitrarily stale model. With use_stale_model=True it blocked until train_iteration caught up to the current ensemble iteration, which is the strict freshness wait that the False case was supposed to perform. Swap them: when a stale model is acceptable, block only until the first model arrives; otherwise block until the current iteration's model is ready. Co-Authored-By: Claude Opus 5 (1M context) --- deepdrivewe/workflows/ddwe.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/deepdrivewe/workflows/ddwe.py b/deepdrivewe/workflows/ddwe.py index c293b91..73fbc5b 100644 --- a/deepdrivewe/workflows/ddwe.py +++ b/deepdrivewe/workflows/ddwe.py @@ -314,21 +314,27 @@ def process_simulation_result(self, result: Result) -> None: # If we have all the simulation results, submit the inference task # using the previous iteration's model if len(self.sim_output) == len(self.ensemble.next_sims): - # We need to wait for the streaming train task to finish - if not self.use_stale_model: + # We need to wait for the first streaming train task to finish + if self.use_stale_model and self.train_output is None: # Wait for the streaming train task to finish - self.logger.info('Waiting for streaming train task to finish') + self.logger.info( + 'Waiting for first streaming train task to finish', + ) while self.train_output is None: time.sleep(10) - elif self.use_stale_model: - self.logger.info('Waiting for streaming train task to finish') + # We need to wait for the next streaming train task to finish + # to get a fresh model + elif not self.use_stale_model: + self.logger.info( + 'Waiting for next streaming train task to finish', + ) while self.train_iteration < self.ensemble.iteration: time.sleep(10) # This should hold (see train_stream_processor) assert self.train_output is not None - # If it's okay to use the stale model, submit the inference task + # Submit the inference task using either a stale or fresh model self.submit_task('inference', self.sim_output, self.train_output) @agent()