Decisions never queue behind other conversations; voice=True waits 1.5 s

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

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()

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

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]