diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b26223a2..ac5d1e63 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -51,7 +51,10 @@ jobs: with: node-version: "24" - run: pip install -e ".[test]" - - run: pytest tests/ -m "not live" --cov=turnstone --cov-report=term-missing --cov-report=xml -q + # -v lists each test id as it starts (pytest prints the nodeid at + # logstart), so a hang names the culprit on the last line instead of + # riding the job timeout with only a trail of "..." dots. + - run: pytest tests/ -m "not live" --cov=turnstone --cov-report=term-missing --cov-report=xml -v - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 if: always() with: @@ -83,7 +86,7 @@ jobs: with: node-version: "24" - run: pip install -e ".[test]" - - run: pytest tests/ -m "not live" --storage-backend=postgresql -q + - run: pytest tests/ -m "not live" --storage-backend=postgresql -v env: TURNSTONE_TEST_PG_URL: postgresql+psycopg://postgres:postgres@localhost:5432/turnstone_test diff --git a/docs/judge.md b/docs/judge.md index f752ed45..296d8754 100644 --- a/docs/judge.md +++ b/docs/judge.md @@ -40,7 +40,7 @@ api_key = "" smart_approvals = false # auto-approve high-confidence "approve" LLM verdicts (opt-in) confidence_threshold = 0.95 # Smart Approvals auto-approve bar (LLM recommendation=approve) max_context_ratio = 0.5 # max % of judge context window for history -timeout = 60.0 # seconds (generous for local models) +timeout = 120.0 # seconds (generous for local models) read_only_tools = true # judge can use read_file/list_directory cancel_on_approval = false # stop judging remaining tool calls once user decides ``` @@ -71,7 +71,7 @@ All fields are optional. The judge is enabled by default; use `enabled = false` --judge / --no-judge Enable/disable (default: enabled) --judge-model MODEL Model for judge --judge-provider PROVIDER Provider for judge ---judge-timeout SECONDS LLM judge timeout (default: 60) +--judge-timeout SECONDS LLM judge timeout (default: 120) --judge-confidence FLOAT Confidence threshold, 0-1 (default: 0.95) ``` @@ -193,9 +193,10 @@ Security hardening blocks access to sensitive paths: ### Timeout -The `timeout` setting (default 60 seconds) is a total budget across all judge -turns. Time is decremented after each LLM call. If the budget expires mid-turn, -the judge attempts to parse whatever partial response is available. +The `timeout` setting (default 120 seconds) applies **per turn**, not as a total +budget across turns — each of the up to 5 turns gets a fresh budget, so a slow +earlier turn doesn't starve later ones. If a turn's budget expires, the judge +attempts to parse whatever partial response is available. --- diff --git a/tests/test_deadline.py b/tests/test_deadline.py new file mode 100644 index 00000000..6750d072 --- /dev/null +++ b/tests/test_deadline.py @@ -0,0 +1,66 @@ +"""Tests for turnstone.core.deadline.run_with_deadline. + +The load-bearing property is the daemon worker: on timeout or cancel the call +is abandoned, and the abandoned thread must be a daemon so it can never block +interpreter exit (the bug that motivated the helper — a non-daemon +ThreadPoolExecutor worker is joined by concurrent.futures' atexit hook). +""" + +from __future__ import annotations + +import threading +import time + +import pytest + +from turnstone.core.deadline import ( + DeadlineCancelledError, + DeadlineExceededError, + run_with_deadline, +) + + +def test_returns_result_on_success() -> None: + assert run_with_deadline(lambda: 42, timeout=1.0) == 42 + + +def test_reraises_callable_exception() -> None: + def boom() -> None: + raise ValueError("upstream failed") + + with pytest.raises(ValueError, match="upstream failed"): + run_with_deadline(boom, timeout=1.0) + + +def test_timeout_returns_promptly_and_abandons_a_daemon_worker() -> None: + # The worker sleeps far past the deadline; the call must return promptly + # via DeadlineExceededError, and the abandoned worker must be a daemon so + # it cannot pin interpreter exit. + start = time.monotonic() + with pytest.raises(DeadlineExceededError): + run_with_deadline(lambda: time.sleep(2.0), timeout=0.2, poll=0.05, thread_name="dl-timeout") + assert time.monotonic() - start < 1.0 + stragglers = [t for t in threading.enumerate() if t.name == "dl-timeout" and not t.daemon] + assert stragglers == [], f"non-daemon worker survived: {stragglers}" + + +def test_cancel_returns_promptly() -> None: + cancel = threading.Event() + + def _fire() -> None: + time.sleep(0.1) + cancel.set() + + threading.Thread(target=_fire, daemon=True).start() + start = time.monotonic() + with pytest.raises(DeadlineCancelledError): + run_with_deadline( + lambda: time.sleep(2.0), + timeout=10.0, + cancel_event=cancel, + poll=0.05, + thread_name="dl-cancel", + ) + assert time.monotonic() - start < 1.0 + stragglers = [t for t in threading.enumerate() if t.name == "dl-cancel" and not t.daemon] + assert stragglers == [], f"non-daemon worker survived: {stragglers}" diff --git a/tests/test_judge.py b/tests/test_judge.py index 0446716d..376cd378 100644 --- a/tests/test_judge.py +++ b/tests/test_judge.py @@ -5,7 +5,6 @@ from __future__ import annotations import json import threading import time -from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any from unittest.mock import MagicMock @@ -120,8 +119,7 @@ class TestVerdictParsing: [{"role": "user", "content": "Run echo hello"}], callback_results.append, ) - # Wait for daemon thread - time.sleep(0.5) + _wait_for(callback_results, 1) assert len(heuristics) == 1 assert heuristics[0].tier == "heuristic" @@ -184,14 +182,12 @@ class TestErrorHandling: provider = _make_mock_provider(side_effect=RuntimeError("API error")) judge = _make_judge(provider) - with ThreadPoolExecutor(max_workers=1) as pool: - result = judge._evaluate_single( - _make_item(), - [{"role": "user", "content": "test"}], - cancel_event=None, - executor=pool, - client=MagicMock(), - ) + result = judge._evaluate_single( + _make_item(), + [{"role": "user", "content": "test"}], + cancel_event=None, + client=MagicMock(), + ) assert result is None def test_provider_error_heuristic_still_returned(self): @@ -209,7 +205,7 @@ class TestErrorHandling: [{"role": "user", "content": "test"}], callback_results.append, ) - time.sleep(0.5) + _wait_for(callback_results, 1) assert len(heuristics) == 1 assert heuristics[0].tier == "heuristic" @@ -233,20 +229,19 @@ class TestErrorHandling: [{"role": "user", "content": "test"}], callback_results.append, ) - time.sleep(0.5) + _wait_for(callback_results, 1) assert len(callback_results) == 1 assert callback_results[0].tier == "llm_fallback" - def test_executor_poison_delivers_fallback(self): - """An _ExecutorPoisonedError (a judge-call timeout poisoning the - single-worker executor) restarts the executor AND still delivers one - fallback for the interrupted item — the twin of the generic-exception - path, and load-bearing for Smart Approvals' batch-completeness wait.""" - from turnstone.core.judge import _ExecutorPoisonedError - + def test_evaluate_single_none_delivers_fallback(self): + """A judge-call timeout now surfaces as ``_evaluate_single`` returning + None (the executor-poison restart dance is gone); the daemon must still + deliver exactly one fallback for that item — Smart Approvals waits on + the full verdict set before gating, so a silently-skipped item would + block that wait until its timeout.""" judge = _make_judge() judge._evaluate_single = MagicMock( # type: ignore[method-assign] - side_effect=_ExecutorPoisonedError() + return_value=None ) callback_results: list[IntentVerdict] = [] judge.evaluate( @@ -254,7 +249,7 @@ class TestErrorHandling: [{"role": "user", "content": "test"}], callback_results.append, ) - time.sleep(0.5) + _wait_for(callback_results, 1) assert len(callback_results) == 1 assert callback_results[0].tier == "llm_fallback" @@ -266,14 +261,12 @@ class TestErrorHandling: result_mock.content = "" judge = _make_judge(provider) - with ThreadPoolExecutor(max_workers=1) as pool: - result = judge._evaluate_single( - _make_item(), - [{"role": "user", "content": "test"}], - cancel_event=None, - executor=pool, - client=MagicMock(), - ) + result = judge._evaluate_single( + _make_item(), + [{"role": "user", "content": "test"}], + cancel_event=None, + client=MagicMock(), + ) assert result is None def test_empty_content_length_stop_no_retry(self): @@ -285,14 +278,12 @@ class TestErrorHandling: result_mock.finish_reason = "length" judge = _make_judge(provider) - with ThreadPoolExecutor(max_workers=1) as pool: - result = judge._evaluate_single( - _make_item(), - [{"role": "user", "content": "test"}], - cancel_event=None, - executor=pool, - client=MagicMock(), - ) + result = judge._evaluate_single( + _make_item(), + [{"role": "user", "content": "test"}], + cancel_event=None, + client=MagicMock(), + ) assert result is None # Should have been called exactly once — no retries assert provider.create_completion.call_count == 1 @@ -404,14 +395,12 @@ class TestMultiTurnToolUse: provider.create_completion.side_effect = [turn1, turn2] judge = _make_judge(provider) - with ThreadPoolExecutor(max_workers=1) as pool: - verdict = judge._evaluate_single( - _make_item(), - [{"role": "user", "content": "test"}], - cancel_event=None, - executor=pool, - client=MagicMock(), - ) + verdict = judge._evaluate_single( + _make_item(), + [{"role": "user", "content": "test"}], + cancel_event=None, + client=MagicMock(), + ) assert verdict is not None assert verdict.tier == "llm" assert provider.create_completion.call_count == 2 @@ -454,14 +443,12 @@ class TestMultiTurnToolUse: ] judge = _make_judge(provider) - with ThreadPoolExecutor(max_workers=1) as pool: - judge._evaluate_single( - _make_item(), - [{"role": "user", "content": "test"}], - cancel_event=None, - executor=pool, - client=MagicMock(), - ) + judge._evaluate_single( + _make_item(), + [{"role": "user", "content": "test"}], + cancel_event=None, + client=MagicMock(), + ) # Should have called create_completion exactly _JUDGE_MAX_TURNS times assert provider.create_completion.call_count == 5 @@ -507,7 +494,7 @@ class TestConfidenceArbitration: [{"role": "user", "content": "Run echo hello"}], callback_results.append, ) - time.sleep(0.5) + _wait_for(callback_results, 1) assert len(heuristics) == 1 assert heuristics[0].confidence == 0.85 @@ -527,7 +514,7 @@ class TestConfidenceArbitration: [{"role": "user", "content": "Run echo hello"}], callback_results.append, ) - time.sleep(0.5) + _wait_for(callback_results, 1) assert len(heuristics) == 1 # LLM verdict is always delivered regardless of confidence comparison @@ -966,11 +953,7 @@ class TestModelAliasResolution: [{"role": "user", "content": "delegate the audit"}], callback_results.append, ) - # Wait for daemon thread. - for _ in range(20): - if callback_results: - break - time.sleep(0.1) + _wait_for(callback_results, 1) assert callback_results, "judge never delivered a verdict" assert callback_results[0].tier == "llm" diff --git a/tests/test_output_guard_judge.py b/tests/test_output_guard_judge.py index 8645b44f..35cee581 100644 --- a/tests/test_output_guard_judge.py +++ b/tests/test_output_guard_judge.py @@ -220,6 +220,26 @@ class TestEvaluateFailurePaths: # Cancel should return promptly, well below the 10s timeout. assert elapsed < 2.0, f"cancel returned in {elapsed:.2f}s, expected < 2.0s" + def test_timeout_leaves_no_nondaemon_straggler(self) -> None: + # Regression: evaluate() abandons a slow upstream call on timeout, but + # the worker must be a *daemon* so it can never pin interpreter exit. + # The old ThreadPoolExecutor worker was non-daemon and got joined by + # concurrent.futures' atexit hook, hanging the whole test run at + # shutdown. See turnstone/core/deadline.py. + judge = _make_judge( + content='{"risk_level":"medium","flags":[],"reasoning":""}', + timeout=1.0, + delay=5.0, + ) + v = judge.evaluate("payload", call_id="c1") + assert v.error == "timeout" + stragglers = [ + t + for t in threading.enumerate() + if t.name.startswith("output-guard-judge") and not t.daemon + ] + assert stragglers == [], f"non-daemon worker survived evaluate(): {stragglers}" + class TestAliasResolution: def test_unknown_alias_falls_back_to_session_model(self) -> None: diff --git a/tests/test_validate_regex_pattern.py b/tests/test_validate_regex_pattern.py new file mode 100644 index 00000000..3ba0a36f --- /dev/null +++ b/tests/test_validate_regex_pattern.py @@ -0,0 +1,38 @@ +"""Tests for turnstone.console.server._validate_regex_pattern. + +The catastrophic-backtracking branch is verified by simulating the deadline +firing rather than running a real ReDoS regex — a genuine runaway pattern would +leave a CPU-pinned daemon worker for the rest of the suite. The daemon-abandon +mechanism itself is covered in tests/test_deadline.py. +""" + +from __future__ import annotations + +from turnstone.console.server import _validate_regex_pattern +from turnstone.core.deadline import DeadlineExceededError + + +def test_valid_pattern_returns_none() -> None: + assert _validate_regex_pattern(r"\d{3}-\d{4}") is None + + +def test_invalid_pattern_returns_error() -> None: + msg = _validate_regex_pattern(r"(unclosed") + assert msg is not None + assert msg.startswith("Invalid regex") + + +def test_catastrophic_backtracking_returns_message(monkeypatch) -> None: + def _deadline(*_args, **_kwargs): + raise DeadlineExceededError + + monkeypatch.setattr("turnstone.console.server.run_with_deadline", _deadline) + assert _validate_regex_pattern(r"(a+)+$") == "Regex appears to have catastrophic backtracking" + + +def test_probe_error_returns_generic_message(monkeypatch) -> None: + def _err(*_args, **_kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr("turnstone.console.server.run_with_deadline", _err) + assert _validate_regex_pattern(r"abc") == "Regex caused an error during test" diff --git a/turnstone/cli.py b/turnstone/cli.py index d1eba511..a22bb513 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -1022,8 +1022,8 @@ def main() -> None: "--judge-timeout", dest="judge_timeout", type=float, - default=60.0, - help="LLM judge timeout in seconds (default: 60)", + default=120.0, + help="LLM judge timeout in seconds (default: 120)", ) judge_group.add_argument( "--judge-confidence", diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 5326a3fd..d86ff09d 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -55,6 +55,7 @@ from turnstone.core.auth import ( jwt_version_slot, require_permission, ) +from turnstone.core.deadline import DeadlineExceededError, run_with_deadline from turnstone.core.rendezvous import NoAvailableNodeError from turnstone.core.session_replay import session_replay_preamble from turnstone.core.session_routes import ( @@ -11530,16 +11531,15 @@ def _validate_regex_pattern(pattern: str, flags: int = 0) -> str | None: compiled.search(s) try: - from concurrent.futures import ThreadPoolExecutor - from concurrent.futures import TimeoutError as FuturesTimeout - - pool = ThreadPoolExecutor(max_workers=1) - try: - pool.submit(_probe).result(timeout=0.5) - except FuturesTimeout: - return "Regex appears to have catastrophic backtracking" - finally: - pool.shutdown(wait=False, cancel_futures=True) + # Daemon worker: a catastrophically-backtracking regex must be + # abandonable without pinning a non-daemon thread that would hang + # interpreter exit (a ThreadPoolExecutor worker is joined at exit). + # Budget is generous — a legitimately complex pattern can take a second + # or two on the probe strings; only exponential blowup (which sails past + # any few-second bound) should trip the catastrophic-backtracking guard. + run_with_deadline(_probe, timeout=3.0, poll=0.1, thread_name="regex-redos-probe") + except DeadlineExceededError: + return "Regex appears to have catastrophic backtracking" except Exception: return "Regex caused an error during test" return None diff --git a/turnstone/core/deadline.py b/turnstone/core/deadline.py new file mode 100644 index 00000000..5afd9aef --- /dev/null +++ b/turnstone/core/deadline.py @@ -0,0 +1,82 @@ +"""Run a blocking call under a wall-clock deadline on a daemon thread. + +The motivating constraint comes from the judges (:mod:`turnstone.core.judge`, +:mod:`turnstone.core.output_guard_judge`): an upstream LLM call must be +*abandonable* the instant its timeout or cancel fires, without the abandoned +call being able to block process or interpreter exit. + +A :class:`~concurrent.futures.ThreadPoolExecutor` worker is **non-daemon**, and +``concurrent.futures`` joins every executor worker from an ``atexit`` hook +(``_python_exit``) regardless of ``shutdown(wait=False)``. So an upstream call +wedged with no socket timeout hangs interpreter shutdown forever — which is +exactly how a single slow judge call can deadlock a whole test run at exit. + +A **daemon** worker is never joined at exit, so abandoning one is always safe: +the call keeps running until it returns or the process dies, whichever comes +first, and never pins shutdown. +""" + +from __future__ import annotations + +import queue +import threading +import time +from typing import TYPE_CHECKING, TypeVar + +if TYPE_CHECKING: + from collections.abc import Callable + +_T = TypeVar("_T") + + +class DeadlineExceededError(Exception): + """The call did not complete before its wall-clock deadline.""" + + +class DeadlineCancelledError(Exception): + """The cancel event fired before the call completed.""" + + +def run_with_deadline( + fn: Callable[[], _T], + *, + timeout: float, + cancel_event: threading.Event | None = None, + poll: float = 1.0, + thread_name: str = "deadline-worker", +) -> _T: + """Run ``fn()`` on a daemon thread, bounded by ``timeout``/``cancel_event``. + + Returns ``fn()``'s result, or re-raises whatever ``fn`` raised. Raises + :class:`DeadlineExceededError` if ``timeout`` seconds elapse first, or + :class:`DeadlineCancelledError` if ``cancel_event`` fires first. On either + abort the worker thread is abandoned; being a daemon it cannot block + process or interpreter exit. + + ``poll`` bounds how often ``cancel_event`` is checked (and thus the worst- + case latency from a cancel to this function returning). + """ + box: queue.Queue[tuple[bool, object]] = queue.Queue(maxsize=1) + + def _runner() -> None: + try: + box.put((True, fn())) + except BaseException as exc: # noqa: BLE001 - relayed to the caller verbatim + box.put((False, exc)) + + threading.Thread(target=_runner, name=thread_name, daemon=True).start() + + deadline = time.monotonic() + timeout + while True: + if cancel_event is not None and cancel_event.is_set(): + raise DeadlineCancelledError + remaining = deadline - time.monotonic() + if remaining <= 0: + raise DeadlineExceededError + try: + ok, payload = box.get(timeout=min(remaining, poll)) + except queue.Empty: + continue + if ok: + return payload # type: ignore[return-value] # ok=True ⇒ payload is _T + raise payload # type: ignore[misc] # ok=False ⇒ payload is the raised exc diff --git a/turnstone/core/judge.py b/turnstone/core/judge.py index d8b3213e..4b456021 100644 --- a/turnstone/core/judge.py +++ b/turnstone/core/judge.py @@ -15,11 +15,16 @@ import re import threading import time import uuid -from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field +from functools import partial from pathlib import Path from typing import TYPE_CHECKING, Any +from turnstone.core.deadline import ( + DeadlineCancelledError, + DeadlineExceededError, + run_with_deadline, +) from turnstone.core.log import get_logger if TYPE_CHECKING: @@ -76,8 +81,8 @@ class JudgeConfig: """Configuration for the intent validation judge. The *timeout* value applies **per turn**, not as a total budget across - all turns. With the default of 60 s and a maximum of 5 turns, a - single tool-call evaluation can take up to 300 s in the worst case + all turns. With the default of 120 s and a maximum of 5 turns, a + single tool-call evaluation can take up to 600 s in the worst case (e.g. a multi-turn tool-use exchange with a slow local model). """ @@ -86,13 +91,13 @@ class JudgeConfig: smart_approvals: bool = False # auto-approve high-confidence "approve" LLM verdicts confidence_threshold: float = 0.95 # Smart Approvals auto-approve bar (recommendation=approve) max_context_ratio: float = 0.5 - timeout: float = 60.0 # per-turn timeout in seconds (see class docstring) + timeout: float = 120.0 # per-turn timeout in seconds (see class docstring) read_only_tools: bool = True output_guard: bool = True output_guard_budget_seconds: float = 30.0 # wall-clock budget for output_guard regex scan output_guard_llm: bool = False # enable LLM stage on tool output (issue #560 mitigation #1) output_guard_model: str = "" # alias for the LLM stage; empty = inherit session model - output_guard_llm_timeout: float = 30.0 # wall-clock budget for the LLM stage + output_guard_llm_timeout: float = 60.0 # wall-clock budget for the LLM stage redact_secrets: bool = True # True = the approval gate's resolution aborts remaining evaluations # (saves inference; undone items degrade to ``llm_fallback`` verdicts @@ -881,10 +886,6 @@ If you used read_file to check a target, cite what you found.""" # --------------------------------------------------------------------------- -class _ExecutorPoisonedError(Exception): - """Raised when a timeout leaves the executor's worker thread stuck.""" - - class IntentJudge: """Session-scoped LLM judge for intent validation. @@ -1054,7 +1055,6 @@ class IntentJudge: verdicts are delivered. """ client = self._create_client() - executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="judge-api") try: for idx, (item, h_verdict) in enumerate(zip(items, heuristic_verdicts, strict=True)): if cancel_event and cancel_event.is_set(): @@ -1071,7 +1071,6 @@ class IntentJudge: item, messages, cancel_event, - executor, client, ) if llm_verdict: @@ -1116,17 +1115,6 @@ class IntentJudge: "judge cancelled before evaluating this call", ) return - except _ExecutorPoisonedError: - executor.shutdown(wait=False, cancel_futures=True) - executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="judge-api") - # Deliver a fallback for the interrupted item so every - # call still gets exactly one verdict. Smart Approvals - # waits on the full set before gating; a silently- - # skipped item would otherwise block that wait until - # its timeout (and the advisory UI would miss a chip). - self._deliver_fallbacks( - [item], [h_verdict], callback, "judge executor restarted" - ) except Exception: log.exception( "Judge evaluation failed for %s", @@ -1134,7 +1122,6 @@ class IntentJudge: ) self._deliver_fallbacks([item], [h_verdict], callback, "judge evaluation error") finally: - executor.shutdown(wait=False, cancel_futures=True) try: if hasattr(client, "close"): client.close() @@ -1172,7 +1159,6 @@ class IntentJudge: item: dict[str, Any], messages: list[dict[str, Any]], cancel_event: threading.Event | None, - executor: ThreadPoolExecutor, client: Any, ) -> IntentVerdict | None: """Run LLM judge for a single tool call. Returns verdict or None.""" @@ -1237,32 +1223,29 @@ class IntentJudge: # models aren't penalised for slow earlier turns. per_call_timeout = max(self._config.timeout, 5.0) # at least 5s try: - future = executor.submit( - self._provider.create_completion, - client=client, - model=self._model, - messages=judge_messages, - tools=None if is_last_turn else tools, - max_tokens=2048, - temperature=0.0, - reasoning_effort="medium", + # Each turn runs on its own daemon worker (1s cancel polling). + # A timeout or cancel abandons the call without pinning a + # non-daemon thread that would block interpreter exit — the old + # single-slot ThreadPoolExecutor left a stuck worker that + # poisoned the pool, which is why the restart dance existed. + result = run_with_deadline( + partial( + self._provider.create_completion, + client=client, + model=self._model, + messages=judge_messages, + tools=None if is_last_turn else tools, + max_tokens=2048, + temperature=0.0, + reasoning_effort="medium", + ), + timeout=per_call_timeout, + cancel_event=cancel_event, + thread_name="judge-api", ) - # Poll in 1s increments so we notice cancellation promptly - # instead of blocking for the full per_call_timeout. - deadline = time.monotonic() + per_call_timeout - while True: - remaining = deadline - time.monotonic() - if cancel_event and cancel_event.is_set(): - future.cancel() - return None - if remaining <= 0: - raise TimeoutError - try: - result = future.result(timeout=min(remaining, 1.0)) - break - except TimeoutError: - pass # loop back to check remaining/cancel - except TimeoutError: + except DeadlineCancelledError: + return None + except DeadlineExceededError: log.info("judge.turn.timeout", turn=turn + 1, timeout=per_call_timeout) # Safety net: if we have a partial result from a previous turn, # try to parse a verdict from it before giving up. @@ -1277,7 +1260,7 @@ class IntentJudge: if verdict: log.info("judge.verdict.from_partial", turn=turn + 1) return verdict - raise _ExecutorPoisonedError from None + return None except Exception as e: log.info("judge.turn.failed", turn=turn + 1, error=str(e)) return None diff --git a/turnstone/core/output_guard_judge.py b/turnstone/core/output_guard_judge.py index f1c68536..96c35507 100644 --- a/turnstone/core/output_guard_judge.py +++ b/turnstone/core/output_guard_judge.py @@ -13,12 +13,12 @@ Design: already in hand. - JSON-in-content verdict. 4-strategy parser inlined from :class:`IntentJudge` (``judge.py:1603-1659``). -- ``ThreadPoolExecutor`` + ``future.result(timeout=)`` with 1 s - cancel-event polling. The executor is owned explicitly with - ``shutdown(wait=False, cancel_futures=True)`` so a timeout or - cancellation returns promptly even if the worker thread is still - blocked on the upstream LLM call. This mirrors - :meth:`IntentJudge._run_judge`'s pattern at ``judge.py:1117-1118``. +- Wall-clock deadline via :func:`turnstone.core.deadline.run_with_deadline`, + which runs the call on a *daemon* worker and polls the cancel event each + second. A timeout or cancel abandons the call rather than waiting it out, + and the daemon worker can never block process or interpreter exit — unlike + a ``ThreadPoolExecutor`` worker, which ``concurrent.futures`` joins from an + ``atexit`` hook regardless of ``shutdown(wait=False)``. - HTTP client is lazy-init + reused across evaluations on a single judge instance. Session-side model swaps drop the entire :class:`OutputGuardJudge` (``session.py:1733``/``:2136``), which @@ -39,11 +39,15 @@ import json import re import time import uuid -from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import TYPE_CHECKING, Any from turnstone.core import fence +from turnstone.core.deadline import ( + DeadlineCancelledError, + DeadlineExceededError, + run_with_deadline, +) from turnstone.core.log import get_logger if TYPE_CHECKING: @@ -370,9 +374,10 @@ class OutputGuardJudge: by :meth:`_user_prompt`. Callers that don't have a particular field leave it at its default — the prompt skips empty sections. - Timeout enforcement is real wall-clock: the executor is shut - down with ``wait=False, cancel_futures=True`` on the timeout / - cancel path, so a hung upstream LLM call does not block return. + Timeout enforcement is real wall-clock: the upstream call runs on a + daemon worker via :func:`~turnstone.core.deadline.run_with_deadline` + and is abandoned on the timeout / cancel path, so a hung upstream LLM + call neither blocks return nor pins interpreter exit. """ if not output: return OutputJudgeVerdict( @@ -407,15 +412,16 @@ class OutputGuardJudge: verdict_id, call_id, start, f"client_create_failed: {type(e).__name__}" ) - # Explicit executor lifetime — the `with ... as ex:` form's - # implicit shutdown(wait=True) would block return until the - # upstream call completed, defeating the wall-clock timeout. - # Mirror IntentJudge's pattern at judge.py:1117-1118. - ex = ThreadPoolExecutor(max_workers=1, thread_name_prefix="output-guard-judge") + # Run the upstream call on a *daemon* worker bounded by a real + # wall-clock deadline: a timeout or cancel abandons the call instead of + # waiting it out, and because the worker is a daemon an abandoned call + # can never block process or interpreter exit. (A ThreadPoolExecutor + # worker is non-daemon, and concurrent.futures joins it from an atexit + # hook regardless of shutdown(wait=False) — so a wedged upstream call + # would otherwise hang shutdown.) try: - try: - future = ex.submit( - self._provider.create_completion, + result = run_with_deadline( + lambda: self._provider.create_completion( client=client, model=self._model, messages=judge_messages, @@ -423,27 +429,19 @@ class OutputGuardJudge: max_tokens=512, temperature=0.0, reasoning_effort="low", - ) - deadline = time.monotonic() + timeout - while True: - if cancel_event is not None and cancel_event.is_set(): - future.cancel() - return self._error_verdict(verdict_id, call_id, start, "cancelled") - remaining = deadline - time.monotonic() - if remaining <= 0: - future.cancel() - return self._error_verdict(verdict_id, call_id, start, "timeout") - try: - result = future.result(timeout=min(remaining, 1.0)) - break - except TimeoutError: - continue - except Exception as e: - return self._error_verdict( - verdict_id, call_id, start, f"provider_error: {type(e).__name__}" - ) - finally: - ex.shutdown(wait=False, cancel_futures=True) + ), + timeout=timeout, + cancel_event=cancel_event, + thread_name="output-guard-judge", + ) + except DeadlineCancelledError: + return self._error_verdict(verdict_id, call_id, start, "cancelled") + except DeadlineExceededError: + return self._error_verdict(verdict_id, call_id, start, "timeout") + except Exception as e: + return self._error_verdict( + verdict_id, call_id, start, f"provider_error: {type(e).__name__}" + ) content = (getattr(result, "content", "") or "").strip() if not content: diff --git a/turnstone/core/settings_registry.py b/turnstone/core/settings_registry.py index fdf47753..7d239f33 100644 --- a/turnstone/core/settings_registry.py +++ b/turnstone/core/settings_registry.py @@ -577,7 +577,7 @@ def _build_registry() -> dict[str, SettingDef]: SettingDef( "judge.timeout", "float", - 60.0, + 120.0, "Judge evaluation timeout in seconds", "judge", min_value=5.0, @@ -641,7 +641,7 @@ def _build_registry() -> dict[str, SettingDef]: SettingDef( "judge.output_guard_llm_timeout", "float", - 30.0, + 60.0, "Wall-clock budget for the output-guard LLM judge call", "judge", min_value=1.0,