feat: a bounded decision, a breaker, fail open on what a compiled answer can't honour

README.md

@@ -29,11 +29,12 @@ with agentcompile.conversation(ticket.id):
 
 ## What happens on each call
 
-1. With a conversation id, the SDK asks AgentCompile for a decision (`/v1/decide`; it waits at most `timeout`, default 2 s).
-2. **compiled**: the answer (a tool call or a reply) comes back in exactly the provider's response shape; your loop runs the tool as usual.
+1. With a conversation id, the SDK asks AgentCompile for a decision (`/v1/decide`). The whole call takes at most `timeout` (default 2 s), and AgentCompile is told that, so it answers in time.
+2. **compiled**: the answer (a tool call or a reply) comes back in exactly the provider's response shape; your loop runs the tool as usual. A tool call is only ever one of the `tools` you passed.
 3. **forwarded**: your model is called as normal.
-4. **fail-open**: AgentCompile errored, was slow or sent something unusable, so your model is called as normal.
-5. **no conversation id**: your model is called and no decision is made.
+4. **fail-open**: AgentCompile errored, was slow or sent something unusable, so your model is called as normal. After 5 failures in a row the SDK stops asking for 30 s, then tries one call.
+5. **unsupported**: the request asks for something a compiled answer couldn't honour (`n` > 1, a forced `tool_choice`, a `response_format`), so your model is called without asking.
+6. **no conversation id**: your model is called and no decision is made.
 
 Your provider key never leaves your process. `mode="shadow"` decides but always calls your model, so you can see what it would have done before you switch it on.
 

src/agentcompile/__init__.py

@@ -15,7 +15,7 @@
 import os
 from typing import Any
 
-from ._capture import Capturer, flush_all
+from ._capture import capturer_for, flush_all
 from ._conversation import conversation
 from ._core import LIVE, SHADOW, Proxy, Router, async_create, sync_create
 from ._decide import AsyncDecider, Decider, Settings
@@ -65,14 +65,15 @@ def wrap(
         key=key or os.environ.get("AGENTCOMPILE_KEY"),
         company=company or os.environ.get("AGENTCOMPILE_COMPANY"),
         timeout=timeout,
+        mode=mode,
     )
     the_trail = Trail(trail, on_event)
     _remember(settings)  # outcome() sends with the client wrapped last
     if capture is None:
         capture = os.environ.get("AGENTCOMPILE_CAPTURE", "") in ("1", "true", "yes")
     # One sender per wrapped client; it always sends from its own thread (sync client).
     capturer = (
-        Capturer(
+        capturer_for(
             settings,
             http_client if _is_sync_http(http_client) else None,
             load_key() if scrub else None,

src/agentcompile/_assemble.py

@@ -101,8 +101,9 @@ def __init__(self, stream: Any, on_done: OnDone) -> None:
         self._on_done = on_done
         self._chunks: list[Any] = []
         self._finished = False
+        self._it = self._iterate()  # one pass, shared by `for` and next()
 
-    def __iter__(self) -> Iterator[Any]:
+    def _iterate(self) -> Iterator[Any]:
         complete = False
         try:
             for item in self._stream:
@@ -112,6 +113,12 @@ def __iter__(self) -> Iterator[Any]:
         finally:
             self._finish(complete=complete)
 
+    def __iter__(self) -> Iterator[Any]:
+        return self._it
+
+    def __next__(self) -> Any:
+        return next(self._it)
+
     def __enter__(self) -> CapturingStream:
         enter = getattr(self._stream, "__enter__", None)
         if enter is not None:
@@ -146,6 +153,7 @@ def __init__(self, stream: Any, on_done: OnDone) -> None:
         self._on_done = on_done
         self._chunks: list[Any] = []
         self._finished = False
+        self._it = self._iterate()
 
     async def _iterate(self) -> AsyncIterator[Any]:
         complete = False
@@ -158,7 +166,10 @@ async def _iterate(self) -> AsyncIterator[Any]:
             self._finish(complete=complete)
 
     def __aiter__(self) -> AsyncIterator[Any]:
-        return self._iterate()
+        return self._it
+
+    async def __anext__(self) -> Any:
+        return await self._it.__anext__()
 
     async def __aenter__(self) -> AsyncCapturingStream:
         enter = getattr(self._stream, "__aenter__", None)

src/agentcompile/_capture.py

@@ -11,6 +11,7 @@
 
 import atexit
 import collections
+import json
 import threading
 import time
 from datetime import datetime, timezone
@@ -25,17 +26,35 @@
 
 MAX_QUEUE = 2000
 BATCH = 50
+# The server takes up to 8 MiB a request: batches stay under 4 MiB, and a call bigger than
+# MAX_RECORD_BYTES on its own (a huge history) is dropped and counted.
+MAX_BATCH_BYTES = 4 * 1024 * 1024
+MAX_RECORD_BYTES = 7 * 1024 * 1024
 INTERVAL = 1.0
 
 
-_ALL: list[Capturer] = []
+_ALL: dict[tuple[Any, ...], Capturer] = {}
+_ALL_LOCK = threading.Lock()
 
 
 def flush_all(timeout: float = 5.0) -> None:
-    for capturer in list(_ALL):
+    for capturer in list(_ALL.values()):
         capturer.flush(timeout)
 
 
+def capturer_for(
+    settings: Settings, http: httpx.Client | None, scrub_key: bytes | None
+) -> Capturer:
+    """One sender (one queue, one thread) per destination: wrapping a client per request, as
+    some apps do, reuses it instead of starting another each time."""
+    key = (settings.base_url, settings.key, settings.company, scrub_key, id(http))
+    with _ALL_LOCK:
+        found = _ALL.get(key)
+        if found is None:
+            found = _ALL[key] = Capturer(settings, http, scrub_key)
+        return found
+
+
 class Capturer:
     def __init__(
         self,
@@ -55,7 +74,6 @@ def __init__(
         self.dropped = 0
         self.failed = 0
         atexit.register(self.flush, 2.0)
-        _ALL.append(self)
 
     def add(
         self,
@@ -144,18 +162,43 @@ def _run(self) -> None:
             while self._queue:
                 self._send_batch()
 
-    def _send_batch(self) -> None:
+    def _take(self) -> list[str]:
+        """Up to BATCH queued calls, encoded, under MAX_BATCH_BYTES together (the first one
+        alone may be up to MAX_RECORD_BYTES); what doesn't fit goes back to the front."""
         with self._lock:
-            batch = [self._queue.popleft() for _ in range(min(BATCH, len(self._queue)))]
+            records = [self._queue.popleft() for _ in range(min(BATCH, len(self._queue)))]
+        batch: list[str] = []
+        size = 0
+        for i, record in enumerate(records):
+            try:
+                line = json.dumps(record, default=str)
+            except Exception:
+                self.dropped += 1
+                continue
+            if len(line) > MAX_RECORD_BYTES:
+                self.dropped += 1
+                continue
+            if batch and size + len(line) > MAX_BATCH_BYTES:
+                with self._lock:  # the rest waits for the next batch, in order
+                    self._queue.extendleft(reversed(records[i:]))
+                break
+            batch.append(line)
+            size += len(line)
+        return batch
+
+    def _send_batch(self) -> None:
+        batch = self._take()
         if not batch:
             return
-        url, headers, _ = _request(self.settings, batch[0]["provider"], "capture", {})
+        url, headers, _ = _request(self.settings, "openai", "capture", {})
         headers.pop(CONVERSATION_HEADER, None)
+        headers["content-type"] = "application/json"
         url = url.rsplit("/v1/decide", 1)[0] + "/v1/capture"
         try:
             if self._http is None:
                 self._http = httpx.Client(timeout=self.settings.timeout)
-            response = self._http.post(url, headers=headers, json={"exchanges": batch})
+            body = '{"exchanges": [' + ", ".join(batch) + "]}"
+            response = self._http.post(url, headers=headers, content=body.encode("utf-8"))
             if response.status_code == 200:
                 self.sent += len(batch)
             else:

src/agentcompile/_core.py

@@ -15,6 +15,46 @@
 LIVE, SHADOW = "live", "shadow"
 
 
+def _given(value: Any) -> Any:
+    """None for an argument left out, however the SDK spells that (NOT_GIVEN, omit)."""
+    return None if type(value).__name__ in ("NotGiven", "Omit") else value
+
+
+def unsupported(kwargs: dict[str, Any]) -> str | None:
+    """Why a compiled answer couldn't honour this request (then it goes to the model, and no
+    decision is asked for): several answers, a forced or forbidden tool, a fixed output
+    shape."""
+    n = _given(kwargs.get("n"))
+    if n is not None and n != 1:
+        return "n"
+    if _given(kwargs.get("functions")) is not None:
+        return "functions"
+    shape = _given(kwargs.get("response_format"))
+    if shape is not None and not (isinstance(shape, dict) and shape.get("type") == "text"):
+        return "response_format"
+    choice = _given(kwargs.get("tool_choice"))
+    auto = choice in (None, "auto") or (
+        isinstance(choice, dict) and choice.get("type") == "auto"
+    )
+    if not auto:
+        return "tool_choice"
+    return None
+
+
+def offered(kwargs: dict[str, Any], tool: str | None) -> bool:
+    """Whether the request offers `tool` (OpenAI's {"function": {"name"}}, or a top-level
+    name)."""
+    for spec in _given(kwargs.get("tools")) or []:
+        spec = spec.model_dump() if hasattr(spec, "model_dump") else spec
+        if not isinstance(spec, dict):
+            continue
+        function = spec.get("function")
+        name = function.get("name") if isinstance(function, dict) else spec.get("name")
+        if name == tool:
+            return True
+    return False
+
+
 class Router:
     """Holds the decision client, the trail and the mode for one wrapped client."""
 
@@ -41,18 +81,27 @@ def prepare(self, kwargs: dict[str, Any]) -> tuple[str | None, dict[str, Any]]:
         return conversation_id, kwargs
 
     def plan(
-        self, conversation_id: str | None, decision: Decision | None, error: str | None
-    ) -> tuple[str, Decision | None]:
-        """(route, the decision to answer with or None to forward)."""
+        self,
+        conversation_id: str | None,
+        decision: Decision | None,
+        error: str | None,
+        kwargs: dict[str, Any] | None = None,
+    ) -> tuple[str, Decision | None, str | None]:
+        """(route, the decision to answer with or None to forward, error text)."""
         if conversation_id is None:
-            return "no-conversation", None
+            return "no-conversation", None, error
+        if kwargs is not None and (why := unsupported(kwargs)) is not None:
+            return "unsupported", None, why
         if decision is None:
-            return "fail-open", None
+            return "fail-open", None, error
         if decision.action == "forward":
-            return "forwarded", None
+            return "forwarded", None, error
         if self.mode == SHADOW:
-            return "shadow", None
-        return "compiled", decision
+            return "shadow", None, error
+        if decision.action == "tool_call" and not offered(kwargs or {}, decision.tool):
+            # A tool the agent didn't offer this turn: its loop couldn't run it.
+            return "fail-open", None, "tool not offered"
+        return "compiled", decision, error
 
     def capture(
         self,
@@ -112,11 +161,11 @@ def create(*args: Any, **kwargs: Any) -> Any:
         started = time.perf_counter()
         conversation_id, real_kwargs = router.prepare(kwargs)
         decision = decide_ms = error = None
-        if conversation_id is not None:
+        if conversation_id is not None and unsupported(real_kwargs) is None:
             decision, decide_ms, error = router.decider.decide(
                 router.provider, conversation_id, payload(real_kwargs)
             )
-        route, answer = router.plan(conversation_id, decision, error)
+        route, answer, error = router.plan(conversation_id, decision, error, real_kwargs)
         stream = bool(real_kwargs.get("stream"))
         if answer is not None:
             try:
@@ -161,11 +210,11 @@ async def create(*args: Any, **kwargs: Any) -> Any:
         started = time.perf_counter()
         conversation_id, real_kwargs = router.prepare(kwargs)
         decision = decide_ms = error = None
-        if conversation_id is not None:
+        if conversation_id is not None and unsupported(real_kwargs) is None:
             decision, decide_ms, error = await router.decider.decide(
                 router.provider, conversation_id, payload(real_kwargs)
             )
-        route, answer = router.plan(conversation_id, decision, error)
+        route, answer, error = router.plan(conversation_id, decision, error, real_kwargs)
         stream = bool(real_kwargs.get("stream"))
         if answer is not None:
             try: