Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ include = [
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["src"]
addopts = "--strict-markers --tb=short -q"

[tool.ruff]
Expand Down
30 changes: 21 additions & 9 deletions src/hawk/workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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."""
Expand Down
15 changes: 13 additions & 2 deletions tests/test_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import asyncio
import time

import pytest

Expand Down Expand Up @@ -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
Expand All @@ -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()
Expand Down
Loading