From 4d0b9028b19ee9fdcaebf227611d36987d50805a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 05:52:02 +0000 Subject: [PATCH 1/2] fix: sync decisions never queue behind other conversations; a call's time runs from its send Sync decision calls shared one process-wide pool of 8 threads, and each call's timeout ran from when it was queued. An agent serving many conversations from one process (32 in our benchmark harness) queued its decisions behind each other whenever the service slowed down: the queued calls ran out of time before they were sent and failed open (30 of 1352 calls in one run, most of them answered quickly once they reached the service). The pool now grows with the calls in flight (up to MAX_DECIDE_THREADS = 256; a thread is started only when none is idle), and a call has its full timeout from when it is sent, the budget the deadline header tells the service. Waiting for a free thread, only past 256 at once, is bounded by the timeout too. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018ruNgP5DUwEkpSrXnv5R8T --- src/agentcompile/_decide.py | 56 +++++++++++++++++---- tests/test_decide_pool.py | 97 +++++++++++++++++++++++++++++++++++++ 2 files changed, 143 insertions(+), 10 deletions(-) create mode 100644 tests/test_decide_pool.py diff --git a/src/agentcompile/_decide.py b/src/agentcompile/_decide.py index 3561a66..2498bba 100644 --- a/src/agentcompile/_decide.py +++ b/src/agentcompile/_decide.py @@ -1,9 +1,12 @@ """The decision call to AgentCompile. Never raises: any problem is a `None` decision, which the wrapper treats as forward (fail open). -Bounded: the whole call (name lookup, connect, send, read) takes at most `timeout`, and the -server is told how long that is so it answers in time. After a few failures in a row the -breaker opens: calls go straight to the model, without asking, until it tries again. +Bounded: the whole call (name lookup, connect, send, read) takes at most `timeout` from when +it is sent, and the server is told how long that is so it answers in time. Sync calls run on +a pool that grows with the calls in flight, so one waits for a thread only past +MAX_DECIDE_THREADS at once (and gives up if none frees within `timeout`). After a few +failures in a row the breaker opens: calls go straight to the model, without asking, until it +tries again. """ from __future__ import annotations @@ -147,21 +150,47 @@ def _read(response: Any, ms: float) -> tuple[Decision | None, float, str | None] return decision, ms, None if decision else "malformed decision" +# Threads sync decision calls may use at once, across every wrapped client in the process. +# The pool grows to it as calls overlap (a thread is started only when none is idle), so an +# agent serving many conversations from one process never queues one conversation's decision +# behind others' (with a fixed 8, 32 conversations in one process failed open whenever the +# service slowed down: the queued calls' time ran out before they were sent). +MAX_DECIDE_THREADS = 256 + _POOL: concurrent.futures.ThreadPoolExecutor | None = None _POOL_LOCK = threading.Lock() def _pool() -> concurrent.futures.ThreadPoolExecutor: - """Where sync decision calls run, so the caller waits at most `timeout` in all (httpx's - timeouts are per phase, and a name lookup has none). A call past its time finishes there, - bounded by httpx's own timeouts, while the caller has moved on.""" + """Where sync decision calls run, so the caller waits a bounded time (httpx's timeouts + are per phase, and a name lookup has none). A call past its time finishes there, bounded + by httpx's own timeouts, while the caller has moved on.""" global _POOL with _POOL_LOCK: if _POOL is None: - _POOL = concurrent.futures.ThreadPoolExecutor(8, "agentcompile-decide") + _POOL = concurrent.futures.ThreadPoolExecutor( + MAX_DECIDE_THREADS, "agentcompile-decide" + ) return _POOL +class _Sent: + """When a pooled call started (was sent): its deadline runs from there.""" + + def __init__(self) -> None: + self.event = threading.Event() + self.at = 0.0 + + def mark(self) -> None: + self.at = time.monotonic() + self.event.set() + + +def _timed(sent: _Sent, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + sent.mark() + return fn(*args, **kwargs) + + class Decider: def __init__(self, settings: Settings, http: httpx.Client | None = None) -> None: self.settings = settings @@ -171,19 +200,26 @@ def __init__(self, settings: Settings, http: httpx.Client | None = None) -> None def decide( self, provider: str, conversation_id: str, payload: dict[str, Any] ) -> tuple[Decision | None, float, str | None]: - """(decision or None, milliseconds, error text).""" + """(decision or None, milliseconds, error text). The call has `timeout` from when it + is sent; waiting for a free thread (only past MAX_DECIDE_THREADS calls at once) is + bounded by `timeout` too.""" if not self.breaker.allow(): return None, 0.0, "circuit open" url, headers, body = _decision_request( self.settings, provider, conversation_id, payload ) + timeout = self.settings.timeout start = time.perf_counter() future = None + sent = _Sent() try: future = _pool().submit( - self._http.post, url, headers=headers, json=body, timeout=self.settings.timeout + _timed, sent, self._http.post, url, headers=headers, json=body, timeout=timeout ) - response = future.result(timeout=self.settings.timeout) + if not sent.event.wait(timeout): + raise concurrent.futures.TimeoutError("no thread free to send it") + left = sent.at + timeout - time.monotonic() + response = future.result(timeout=max(0.0, left)) except concurrent.futures.TimeoutError: if future is not None: future.cancel() diff --git a/tests/test_decide_pool.py b/tests/test_decide_pool.py new file mode 100644 index 0000000..d7a09ad --- /dev/null +++ b/tests/test_decide_pool.py @@ -0,0 +1,97 @@ +"""Many conversations in one process: sync decision calls never wait behind each other, and a +call's time runs from when it is sent (agentcompile/_decide.py).""" + +from __future__ import annotations + +import concurrent.futures +import threading +import time +from collections.abc import Iterator + +import httpx +import pytest + +from agentcompile import _decide +from agentcompile._decide import Decider, Settings + +SAY = {"action": "say", "text": "done"} +MESSAGES = [{"role": "user", "content": "cancel order #W1"}] + + +def _slow(seconds: float) -> httpx.Client: + def handler(request: httpx.Request) -> httpx.Response: + time.sleep(seconds) + return httpx.Response(200, json=SAY) + + return httpx.Client(transport=httpx.MockTransport(handler)) + + +@pytest.fixture +def pool_of(monkeypatch: pytest.MonkeyPatch) -> Iterator[object]: + """Swap the process's decision pool for one of a given size.""" + made: list[concurrent.futures.ThreadPoolExecutor] = [] + + def use(size: int) -> concurrent.futures.ThreadPoolExecutor: + pool = concurrent.futures.ThreadPoolExecutor(size, "test-decide") + made.append(pool) + monkeypatch.setattr(_decide, "_POOL", pool) + return pool + + yield use + for pool in made: + pool.shutdown(wait=False, cancel_futures=True) + + +def test_thirty_two_conversations_at_once_all_get_their_decision( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The service takes 0.3 s; the budget is 1 s. With 8 threads the last calls waited + 1.2 s for a thread and failed open; the pool now grows with the calls in flight.""" + monkeypatch.setattr(_decide, "_POOL", None) + decider = Decider(Settings("http://ac.test", "k", None, 1.0), _slow(0.3)) + start = threading.Barrier(32) + + def one(i: int) -> str | None: + start.wait() + _, _, error = decider.decide("openai", f"c{i}", {"messages": MESSAGES}) + return error + + with concurrent.futures.ThreadPoolExecutor(32) as callers: + errors = list(callers.map(one, range(32))) + assert errors == [None] * 32 + assert _decide.MAX_DECIDE_THREADS >= 32 + + +def test_the_time_runs_from_when_the_call_is_sent(pool_of: object) -> None: + pool_of(1) # type: ignore[operator] + busy = Decider(Settings("http://ac.test", "k", None, 1.0), _slow(0.3)) + waiting = Decider(Settings("http://ac.test", "k", None, 0.5), _slow(0.3)) + first = threading.Thread(target=busy.decide, args=("openai", "a", {"messages": MESSAGES})) + first.start() + time.sleep(0.05) + # Waits ~0.25 s for the one thread, then the call itself takes 0.3 s: 0.55 s in all, + # past 0.5 s counted from the queue, within it counted from the send. + decision, _, error = waiting.decide("openai", "b", {"messages": MESSAGES}) + first.join() + assert error is None and decision is not None and decision.text == "done" + + +def test_a_call_no_thread_frees_up_for_fails_open_within_its_time(pool_of: object) -> None: + pool = pool_of(1) # type: ignore[operator] + release = threading.Event() + pool.submit(release.wait, 5) # the only thread, held + decider = Decider(Settings("http://ac.test", "k", None, 0.2), _slow(0.0)) + started = time.perf_counter() + decision, _, error = decider.decide("openai", "c", {"messages": MESSAGES}) + release.set() + assert decision is None and error == "deadline" + assert time.perf_counter() - started < 0.5 + + +def test_a_slow_call_once_sent_still_gives_up_at_its_time(pool_of: object) -> None: + pool_of(4) # type: ignore[operator] + decider = Decider(Settings("http://ac.test", "k", None, 0.2), _slow(1.0)) + started = time.perf_counter() + decision, _, error = decider.decide("openai", "c", {"messages": MESSAGES}) + assert decision is None and error == "deadline" + assert time.perf_counter() - started < 0.5 From b3a6240df58f85edc61fb0532033ace40a7cb9d6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 16:29:57 +0000 Subject: [PATCH 2/2] feat: voice=True waits 1.5 s for a decision before failing open A voice caller hears every second of waiting. wrap(..., voice=True) sets the default decision timeout to 1.5 s (chat stays 5 s); an explicit timeout still wins. Decisions take 0.6 s at the median and 1.5 s at p90 in a 1,156-turn voice benchmark. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018ruNgP5DUwEkpSrXnv5R8T --- src/agentcompile/__init__.py | 15 +++++++++++++-- tests/test_wrap.py | 22 ++++++++++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/src/agentcompile/__init__.py b/src/agentcompile/__init__.py index 2495264..fac839b 100644 --- a/src/agentcompile/__init__.py +++ b/src/agentcompile/__init__.py @@ -59,32 +59,43 @@ DEFAULT_URL = "https://api.tryagentcompile.com" +# Seconds a decision may take before the call fails open to the wrapped model. A voice +# caller hears the wait; our decisions take 0.6 s at the median and 1.5 s at p90. +CHAT_TIMEOUT = 5.0 +VOICE_TIMEOUT = 1.5 + + def wrap( client: Any, *, key: str | None = None, base_url: str | None = None, mode: str = LIVE, - timeout: float = 5.0, + timeout: float | None = None, trail: str | bool | None = True, on_event: OnEvent | None = None, company: str | None = None, http_client: Any = None, capture: bool | None = None, scrub: bool = True, + voice: bool = False, ) -> Any: """Wrap an OpenAI- or Anthropic-style client; returns an object you use exactly like it. key: your AgentCompile key (or AGENTCOMPILE_KEY). base_url: AGENTCOMPILE_URL or the default. mode: "live" answers known jobs; "shadow" decides but always calls your model, so you can - see what it would have done. timeout: seconds to wait for a decision before failing open. + see what it would have done. timeout: seconds to wait for a decision before failing open + (default 5, or 1.5 with voice=True: a caller hears every second of waiting). trail: a path, True for ~/.agentcompile/trail.jsonl, or False. on_event: called per call. capture: send each call's request and answer to AgentCompile in the background, so it can find the jobs your agent repeats (opt-in; or AGENTCOMPILE_CAPTURE=1). Never slows a call. scrub: with capture, turn emails, cards, phones and account numbers into keyed tokens on this machine before anything is sent (default True; the key is AGENTCOMPILE_SCRUB_KEY or one created once in ~/.agentcompile/). + voice: the agent talks to callers (a voice agent); only sets the default timeout. """ + if timeout is None: + timeout = VOICE_TIMEOUT if voice else CHAT_TIMEOUT settings = _settings(key, base_url, company, timeout, mode) the_trail = Trail(trail, on_event) _remember(settings) # outcome() sends with the client wrapped last diff --git a/tests/test_wrap.py b/tests/test_wrap.py index 25f960f..306acd4 100644 --- a/tests/test_wrap.py +++ b/tests/test_wrap.py @@ -366,3 +366,25 @@ def test_version_flag(capsys: pytest.CaptureFixture[str]) -> None: main(["--version"]) assert done.value.code == 0 assert capsys.readouterr().out.startswith("agentcompile ") + + +@pytest.mark.parametrize( + ("kwargs", "expected"), + [({}, 5.0), ({"voice": True}, 1.5), ({"voice": True, "timeout": 3.0}, 3.0)], +) +def test_a_voice_agent_waits_less_for_a_decision( + monkeypatch: pytest.MonkeyPatch, kwargs: dict[str, Any], expected: float +) -> None: + # A caller hears every second of waiting: a voice agent fails open after 1.5 s unless + # it names its own timeout. + seen: list[float] = [] + real = agentcompile._settings + + def spy(key: Any, base_url: Any, company: Any, timeout: float, mode: str) -> Any: + seen.append(timeout) + return real(key, base_url, company, timeout, mode) + + monkeypatch.setattr(agentcompile, "_settings", spy) + client = openai.OpenAI(api_key="sk-test", base_url="http://provider.invalid/v1") + agentcompile.wrap(client, key="ac-test", trail=False, **kwargs) + assert seen == [expected]