diff --git a/pyproject.toml b/pyproject.toml index bd63ba6..53c419b 100644 --- a/pyproject.toml +++ b/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] diff --git a/src/hawk/workflow.py b/src/hawk/workflow.py index 23cd073..1cf0353 100644 --- a/src/hawk/workflow.py +++ b/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.""" diff --git a/tests/test_workflow.py b/tests/test_workflow.py index c5fcc24..cf4f17b 100644 --- a/tests/test_workflow.py +++ b/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()