Make Windows qualification cleanup deterministic

tests/test_phase2_recipe_coordinator_qualification.py

@@ -7,6 +7,34 @@
 from tools.execution_spikes import recipe_coordinator_e2e_qualification as qualification
 
 
+def test_bounded_cleanup_returns_action_errors_without_thread_traceback():
+    error = qualification._bounded_cleanup(
+        lambda: (_ for _ in ()).throw(RuntimeError("cleanup failed")),
+    )
+
+    assert isinstance(error, RuntimeError)
+
+
+def test_workspace_cleanup_retries_transient_windows_errors(monkeypatch, tmp_path: Path):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    attempts = 0
+    original_rmtree = qualification.shutil.rmtree
+
+    def flaky_rmtree(path):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise PermissionError("native handle still closing")
+        return original_rmtree(path)
+
+    monkeypatch.setattr(qualification.shutil, "rmtree", flaky_rmtree)
+    qualification._remove_workspace(workspace, timeout_seconds=1)
+
+    assert attempts == 2
+    assert not workspace.exists()
+
+
 def test_coordinator_probe_blocks_without_native_windows(monkeypatch, tmp_path: Path):
     monkeypatch.setattr(qualification.os, "name", "posix")
 

tools/execution_spikes/recipe_coordinator_e2e_qualification.py

@@ -310,6 +310,27 @@ def _native_cleanup(
     _require(not leaked, "native_worker_process_leaked")
 
 
+def _remove_workspace(workspace: Path, *, timeout_seconds: float = 10.0) -> None:
+    """Retry temporary-tree removal while native Windows handles settle."""
+
+    deadline = time.monotonic() + timeout_seconds
+    last_error: OSError | None = None
+    while workspace.exists():
+        try:
+            shutil.rmtree(workspace)
+        except FileNotFoundError:
+            return
+        except OSError as error:
+            last_error = error
+        if not workspace.exists():
+            return
+        if time.monotonic() >= deadline:
+            if last_error is not None:
+                raise last_error
+            raise OSError("qualification workspace cleanup timed out")
+        time.sleep(0.05)
+
+
 def qualify(
     source_root: Path,
     *,
@@ -426,7 +447,10 @@ def process_factory_factory() -> _RecordingProcessFactory:
                 stages.append("native_worker_processes_closed")
             except QualificationFailure as error:
                 cleanup_error = error.code
-        _bounded_cleanup(lambda: shutil.rmtree(workspace))
+        _bounded_cleanup(
+            lambda: _remove_workspace(workspace),
+            timeout_seconds=15.0,
+        )
         if workspace.exists():
             cleanup_error = cleanup_error or "qualification_workspace_cleanup_failed"
 

tools/execution_spikes/recipe_worker_e2e_qualification.py

@@ -234,12 +234,23 @@ def _response_code(message: BrokerMessage, operation: str) -> str:
     return str(message.body.get("schema_version", ""))
 
 
-def _bounded_cleanup(action: Any, timeout_seconds: float = 5.0) -> None:
+def _bounded_cleanup(action: Any, timeout_seconds: float = 5.0) -> BaseException | None:
+    """Run cleanup without leaking an exception from the daemon thread.
+
+    Native qualification cleanup can briefly race process-handle teardown on
+    Windows.  The caller still decides whether the cleanup completed, but an
+    exception must be returned to the caller instead of being printed by
+    ``threading`` as an unrelated traceback.
+    """
+
     finished = Queue(maxsize=1)
+    failure: list[BaseException] = []
 
     def run() -> None:
         try:
             action()
+        except BaseException as exc:  # pragma: no cover - platform-specific cleanup
+            failure.append(exc)
         finally:
             finished.put(True)
 
@@ -248,7 +259,8 @@ def run() -> None:
     try:
         finished.get(timeout=timeout_seconds)
     except Exception:
-        pass
+        return None
+    return failure[0] if failure else None
 
 
 def _install_ephemeral(source_root: Path, store_root: Path) -> SignedBundleInstaller: