feat: retry failed capture batches with backoff; default decision budget 5 s

README.md

@@ -29,7 +29,7 @@ with agentcompile.conversation(ticket.id):
 
 ## What happens on each call
 
-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.
+1. With a conversation id, the SDK asks AgentCompile for a decision (`/v1/decide`). The whole call takes at most `timeout` (default 5 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. After 5 failures in a row the SDK stops asking for 30 s, then tries one call.

src/agentcompile/__init__.py

@@ -65,7 +65,7 @@ def wrap(
     key: str | None = None,
     base_url: str | None = None,
     mode: str = LIVE,
-    timeout: float = 2.0,
+    timeout: float = 5.0,
     trail: str | bool | None = True,
     on_event: OnEvent | None = None,
     company: str | None = None,
@@ -145,7 +145,7 @@ def transport(
     base_url: str | None = None,
     company: str | None = None,
     scrub: bool = True,
-    timeout: float = 2.0,
+    timeout: float = 5.0,
 ) -> Any:
     """A transport that captures every model call through it, for frameworks whose client you
     don't hold (LangChain, LiteLLM, CrewAI, Pydantic AI, ...). Capture only: calls are never
@@ -165,7 +165,7 @@ def async_transport(
     base_url: str | None = None,
     company: str | None = None,
     scrub: bool = True,
-    timeout: float = 2.0,
+    timeout: float = 5.0,
 ) -> Any:
     """The async version of transport()."""
     _, async_ = classes(lib_of(wrapped, lib))

src/agentcompile/_capture.py

@@ -2,8 +2,10 @@
 can find the jobs your agent repeats (opt-in: `wrap(..., capture=True)`).
 
 Never in the way: calls are queued and sent by one background thread in batches; a full queue
-drops the oldest, a failed send is dropped and counted, and nothing here ever raises into your
-agent. Each record is one line in the exchange-log shape AgentCompile's importer reads:
+drops the oldest, a send that fails for a passing reason (no connection, a timeout, 408, 429,
+5xx) is retried a few times with backoff, one that still fails is dropped and counted, and
+nothing here ever raises into your agent. Each record is one line in the exchange-log shape
+AgentCompile's importer reads:
 {"provider", "conversation_id", "timestamp", "request", "response"}.
 """
 
@@ -31,6 +33,12 @@
 MAX_BATCH_BYTES = 4 * 1024 * 1024
 MAX_RECORD_BYTES = 7 * 1024 * 1024
 INTERVAL = 1.0
+# A failed batch is tried RETRIES more times, waiting BACKOFF, then twice that, ... (or what
+# the server's Retry-After asks, up to MAX_WAIT); about 3.5 s in all before it is dropped.
+RETRIES = 3
+BACKOFF = 0.5
+MAX_WAIT = 10.0
+_PASSING = frozenset({408, 425, 429, 500, 502, 503, 504})
 
 
 _ALL: dict[tuple[Any, ...], Capturer] = {}
@@ -72,7 +80,10 @@ def __init__(
         self._thread: threading.Thread | None = None
         self.sent = 0
         self.dropped = 0
-        self.failed = 0
+        self.failed = (
+            0  # calls dropped after their send failed (and was retried, if it could be)
+        )
+        self.retried = 0  # batch sends tried again
         atexit.register(self.flush, 2.0)
 
     def add(
@@ -147,7 +158,7 @@ def flush(self, timeout: float = 5.0) -> None:
         """Send everything queued now (tests, and at exit)."""
         deadline = time.monotonic() + timeout
         while self._queue and time.monotonic() < deadline:
-            self._send_batch()
+            self._send_batch(deadline)
 
     def _start(self) -> None:
         if self._thread is None or not self._thread.is_alive():
@@ -195,22 +206,45 @@ def _put_back(self, rest: list[dict[str, Any]]) -> None:
             self.dropped += len(rest) - len(keep)
             self._queue.extendleft(reversed(keep))
 
-    def _send_batch(self) -> None:
+    def _send_batch(self, deadline: float | None = None) -> None:
+        """Send one batch, retrying a passing failure with backoff; never past `deadline` (a
+        flush's), when there is one."""
         batch = self._take()
         if not batch:
             return
         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"
+        body = ('{"exchanges": [' + ", ".join(batch) + "]}").encode("utf-8")
+        for attempt in range(RETRIES + 1):
+            wait = self._post(url, headers, body)
+            if wait is None:
+                self.sent += len(batch)
+                return
+            if wait < 0 or attempt == RETRIES:
+                break
+            wait = min(max(wait, BACKOFF * 2**attempt), MAX_WAIT)
+            if deadline is not None and time.monotonic() + wait > deadline:
+                break
+            self.retried += 1
+            time.sleep(wait)
+        self.failed += len(batch)
+
+    def _post(self, url: str, headers: dict[str, str], body: bytes) -> float | None:
+        """None when the batch was taken; else how long the server asks us to wait before
+        trying again (0: no preference), or -1 for a failure retrying won't fix."""
         try:
             if self._http is None:
                 self._http = httpx.Client(timeout=self.settings.timeout)
-            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:
-                self.failed += len(batch)
+            response = self._http.post(url, headers=headers, content=body)
         except Exception:
-            self.failed += len(batch)
+            return 0.0
+        if response.status_code == 200:
+            return None
+        if response.status_code not in _PASSING:
+            return -1.0
+        try:
+            return float(response.headers.get("retry-after", 0))
+        except ValueError:
+            return 0.0

tests/test_capture.py

@@ -120,6 +120,72 @@ def test_a_failing_service_never_breaks_the_call() -> None:
     )
 
 
+def _capturer_against(handler: Any) -> Any:
+    settings = agentcompile.Settings("http://ac.test", "ack_acme.k", None, 1.0)
+    capturer = _capture.Capturer(settings, httpx.Client(transport=httpx.MockTransport(handler)))
+    capturer._start = lambda: None  # type: ignore[method-assign]
+    capturer.add("openai", "c1", {"model": "gpt-x", "messages": MESSAGES}, COMPLETION)
+    return capturer
+
+
+@pytest.fixture
+def _no_waiting(monkeypatch: pytest.MonkeyPatch) -> list[float]:
+    waits: list[float] = []
+    monkeypatch.setattr(_capture.time, "sleep", waits.append)
+    return waits
+
+
+def test_a_passing_failure_is_retried_until_it_lands(_no_waiting: list[float]) -> None:
+    answers = iter([503, 429, 200])
+    attempts: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if not attempts:
+            attempts.append(0)
+            raise httpx.ConnectError("down")
+        attempts.append(1)
+        return httpx.Response(next(answers), json={})
+
+    capturer = _capturer_against(handler)
+    capturer._send_batch()
+    assert (capturer.sent, capturer.failed, capturer.retried) == (1, 0, 3)
+    assert _no_waiting == [0.5, 1.0, 2.0]  # backoff doubles
+
+
+def test_retries_are_bounded_and_the_drop_counted(_no_waiting: list[float]) -> None:
+    calls: list[int] = []
+    capturer = _capturer_against(lambda r: calls.append(1) or httpx.Response(502))
+    capturer._send_batch()
+    assert len(calls) == _capture.RETRIES + 1
+    assert (capturer.sent, capturer.failed) == (0, 1)
+
+
+def test_a_failure_retrying_wont_fix_is_not_retried(_no_waiting: list[float]) -> None:
+    calls: list[int] = []
+    capturer = _capturer_against(lambda r: calls.append(1) or httpx.Response(401))
+    capturer._send_batch()
+    assert len(calls) == 1 and capturer.failed == 1 and _no_waiting == []
+
+
+def test_retry_after_is_honoured_up_to_a_cap(_no_waiting: list[float]) -> None:
+    answers = iter(
+        [
+            httpx.Response(429, headers={"retry-after": "3"}),
+            httpx.Response(429, headers={"retry-after": "3600"}),
+            httpx.Response(200),
+        ]
+    )
+    capturer = _capturer_against(lambda r: next(answers))
+    capturer._send_batch()
+    assert _no_waiting == [3.0, _capture.MAX_WAIT] and capturer.sent == 1
+
+
+def test_a_flush_never_waits_past_its_deadline(_no_waiting: list[float]) -> None:
+    capturer = _capturer_against(lambda r: httpx.Response(503))
+    capturer.flush(0.1)  # the first backoff (0.5 s) would overrun it
+    assert _no_waiting == [] and capturer.failed == 1
+
+
 def test_a_full_queue_drops_the_oldest(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(_capture, "MAX_QUEUE", 2)
     settings = agentcompile.Settings("http://ac.test", "ack_acme.k", None, 1.0)