fix: drop late reports into an already-finished trial (#996 follow-up)

thinkall's fourth CHANGES_REQUESTED review, point 2: a _RunContext
captured for a trial stays usable after that trial finishes. A worker
thread or executor task that captured get_run_context() and reports
late, after the trial is already TERMINATED, still wrote through:
process_trial_result() overwrote the trial's final metric_analysis and
last_result with the stale value, and report()'s own trailing
`if trial.is_finished(): raise StopIteration` (the normal
scheduler-stop signal for the current report) then raised into the
late caller too, which has no reason to expect it the way a
trainable's own control-flow loop does.

report() now checks trial.is_finished() before processing and drops
the late report instead. New regression test captures a context,
lets the trial finish, then reports through the stale context: fails
with an uncaught StopIteration on the prior commit, passes now, and
asserts last_result/metric_analysis are byte-for-byte unchanged by
the late write.

Point 1 (a persistent callback/queue worker started before tune.run()
never gets context for later items) is not fixed here: replied on the
PR with why an automatic fallback is not safe to ship.

Author: AmirF194

flaml/tune/tune.py

@@ -574,6 +574,15 @@ def compute_with_config(config):
     trial = running_trial if running_trial is not None else getattr(runner, "running_trial", None)
     if not trial:
         return None
+    if trial.is_finished():
+        # A late report from a background thread or executor task whose
+        # captured _RunContext outlived its trial (#996 follow-up, fourth
+        # review point 2): the trial's final result is already recorded,
+        # and process_trial_result() would overwrite it with this stale
+        # value, plus the is_finished() check below would then raise
+        # StopIteration into a caller that never expected it (unlike the
+        # trainable's own control-flow loop, which does). Drop it instead.
+        return None
     result["training_iteration"] = _next_training_iteration(trial)
     result["config"] = trial.config
     if INCUMBENT_RESULT in result["config"]:

test/tune/test_concurrent_run.py

@@ -569,6 +569,48 @@ def worker():
     )
 
 
+def test_late_report_via_stale_context_does_not_corrupt_finished_trial():
+    """Follow-up to #996, fourth review point 2: a _RunContext captured for
+    a trial stays usable after that trial finishes. A worker thread that
+    captured get_run_context() and reports late, after tune.run() has
+    already returned and the trial is TERMINATED, used to still write
+    through: process_trial_result() overwrote the trial's already-final
+    metric_analysis/last_result with the late value, and report()'s own
+    trailing `if trial.is_finished(): raise StopIteration` (the normal
+    scheduler-stop signal for the CURRENT report) then raised into the late
+    caller too, which has no reason to expect it the way a trainable's own
+    control-flow loop does.
+    """
+    captured = {}
+
+    def eval_capturing(config):
+        captured["ctx"] = tune.get_run_context()
+        return {"metric": 1.0}
+
+    analysis = tune.run(
+        eval_capturing,
+        config={"x": tune.uniform(0, 1)},
+        metric="metric",
+        mode="min",
+        num_samples=1,
+        verbose=0,
+    )
+    trial = analysis.trials[0]
+    assert trial.is_finished(), "expected the trial to be TERMINATED once tune.run() returns"
+    last_result_before = dict(trial.last_result)
+    metric_analysis_before = {k: dict(v) for k, v in trial.metric_analysis.items()}
+
+    with tune.use_run_context(captured["ctx"]):
+        tune.report(metric=999.0)
+
+    assert (
+        trial.last_result == last_result_before
+    ), f"a late report corrupted the finished trial's last_result: {trial.last_result}"
+    assert (
+        trial.metric_analysis == metric_analysis_before
+    ), f"a late report corrupted the finished trial's metric_analysis: {trial.metric_analysis}"
+
+
 def test_tune_log_records_from_executor_worker_reach_run_log(tmp_path):
     """Same as above (third review point 4), for a task submitted to a
     ThreadPoolExecutor rather than a plain Thread, since the two are