fix: clean up workflow timeout handling

Author: Patel230

pyproject.toml

@@ -68,6 +68,7 @@ include = [
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]
+pythonpath = ["src"]
 addopts = "--strict-markers --tb=short -q"
 
 [tool.ruff]

src/hawk/workflow.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import asyncio
-import concurrent.futures
+import queue
+import threading
 from dataclasses import dataclass
 from typing import Any, Callable, TypeVar
 
@@ -101,20 +102,31 @@ def run(self, initial_input: Any = None) -> Any:
     def _run_step_with_timeout(self, s: Step, input_val: Any) -> Any:
         """Run a step in a worker thread with a timeout.
 
-        Uses a ThreadPoolExecutor so the timeout works on any thread
+        Uses a daemon worker thread so the timeout works on any thread
         (not just the main thread) and on all platforms (no SIGALRM).
 
         Note: the worker thread keeps running after a timeout; only the
-        caller is unblocked.  This is inherent to cooperative cancellation
-        in Python and matches asyncio.wait_for semantics.
+        caller is unblocked. The thread is marked daemon so a timed-out step
+        cannot keep process shutdown waiting after the caller has moved on.
         """
         assert s.timeout is not None
-        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-            future = pool.submit(self._run_step_sync, s, input_val)
+        result_queue: queue.Queue[tuple[bool, Any]] = queue.Queue(maxsize=1)
+
+        def runner() -> None:
             try:
-                return future.result(timeout=s.timeout)
-            except concurrent.futures.TimeoutError:
-                raise TimeoutError(f"Step '{s.name}' timed out after {s.timeout}s") from None
+                result_queue.put((True, self._run_step_sync(s, input_val)))
+            except Exception as exc:
+                result_queue.put((False, exc))
+
+        worker = threading.Thread(target=runner, name=f"hawk-workflow-{s.name}", daemon=True)
+        worker.start()
+        try:
+            ok, payload = result_queue.get(timeout=s.timeout)
+        except queue.Empty:
+            raise TimeoutError(f"Step '{s.name}' timed out after {s.timeout}s") from None
+        if ok:
+            return payload
+        raise payload
 
     def _run_step_sync(self, s: Step, input_val: Any) -> Any:
         """Run a single step with optional retry."""

tests/test_workflow.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import time
 
 import pytest
 
@@ -97,8 +98,6 @@ def slow(x: int) -> int:
             wf.run(1)
 
     def test_timeout_includes_step_name(self) -> None:
-        import time
-
         def slow(x: int) -> int:
             time.sleep(30.0)
             return x
@@ -107,6 +106,18 @@ def slow(x: int) -> int:
         with pytest.raises(TimeoutError, match="my_step"):
             wf.run(1)
 
+    def test_timeout_returns_promptly(self) -> None:
+        def slow(x: int) -> int:
+            time.sleep(30.0)
+            return x
+
+        wf = Workflow("prompt-timeout").step("slow", slow, timeout=0.5).build()
+        started = time.perf_counter()
+        with pytest.raises(TimeoutError, match="timed out"):
+            wf.run(1)
+        elapsed = time.perf_counter() - started
+        assert elapsed < 2.0
+
     def test_no_timeout_when_none(self) -> None:
         """Steps without a timeout run to completion."""
         wf = Workflow("no-timeout").step("fast", lambda x: x + 1).build()