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/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 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]