feat(api,clients): admission control with a bounded wait, and client retry (#4253)

* feat(api,clients): admission control with a bounded wait, and client retry

The engine already caps concurrent work (`recall_max_concurrent` and friends),
but `async with semaphore` is backpressure, not admission control: it bounds how
much runs and lets an unbounded queue form behind it. Measured on a 2-vCPU
container, 1024 concurrent recalls against a 32-permit semaphore produced a 12.8s
p50 -- the latency did not go away, it moved into the semaphore queue, and the
server spent CPU on responses whose callers had long since gone.

Server: per-operation lanes with a deadline
-------------------------------------------
`api/admission.py` adds lanes with two numbers: how much runs concurrently, and
how long a request may queue before it is refused with 503 + `Retry-After`.

Enforced by an `admit_for` dependency on the same routes that carry
`precheck_for`, so a refusal happens before the body is deserialised -- the
cheapest point to say no, and where an extension already rejects on quota.

Lanes are per operation because per-request cost spans three orders of magnitude
(measured: ~0.1ms /health/live, ~0.5ms bank stats, ~23ms recall); one global cap
calibrated for recall would throttle health checks 100x too hard. Only recall,
reflect and retain get lanes -- the other PrecheckOperations are low-volume
administrative routes, and gating them would add knobs nobody tunes.

Limits are PER WORKER and derived from the CPU budget this process actually has,
via the existing cgroup-aware detector (`os.cpu_count()` reports the host's cores
under `--cpus`, which would size limits for a machine the process cannot use).
`in_flight` is a latency target, not a capacity limit: a c=1024 sweep measured
throughput flat at 40-45 rps whether the limit was 8, 16 or 24 per worker, while
p50 moved 1.4s -> 2.3s. It is bounded on both sides -- too low throttles I/O-bound
work (8 permits against a 500ms provider caps a worker at 16 rps), too high
rebuilds the queue this exists to prevent.

A queued request whose client disconnects releases its place immediately, using
the token `ClientDisconnectCancellationMiddleware` already puts on the scope.
That is what makes a patient 30s deadline affordable: the queue self-cleans, so
waiting costs nothing when nobody is listening. Verified end to end -- a client
that gave up at 1s freed its slot at 1.003s, not at the deadline.

Clients: retry the idempotent calls
-----------------------------------
Both maintained wrappers retry recall and reflect on 429/503. Writes are not
retried: the Python wrapper documents that `operation_id` is ignored for
synchronous retain, so a retry there could duplicate.

Two properties matter more than the retry. `Retry-After` is honoured, because the
server sends it knowing its own queue depth. And the wait is jittered -- a burst
that all receive `Retry-After: 1` and obey it exactly returns in lockstep and
rebuilds the spike. The generated Python client ships `ExponentialRetry`, which
has neither, and was off by default; this is why it stays off.

* fix(admission): decrement queued once on abandon, drop no-op lanes, regen docs skill

- An abandoned waiter decremented stats.queued in its except branch and again in
  finally, driving the gauge negative.
- admit_for on dry-run/mental-model/files routes was a no-op (no lane exists).
- Stale config comments on the kill switch and reflect sizing.
- HTTP-level test for 503 + Retry-After; regenerated docs skill.

* chore(embed): re-sync bundled env.example with repo root

.env.example

@@ -464,6 +464,25 @@ HINDSIGHT_API_LOG_LEVEL=info
 # Explicit refreshes always run immediately.
 # HINDSIGHT_API_MENTAL_MODEL_MIN_REFRESH_INTERVAL_SECONDS=0
 
+# Admission control. The *_MAX_CONCURRENT caps above bound how much runs at once and
+# let an unbounded queue form behind them; these bound how long a request may WAIT
+# before it is refused with 503 + Retry-After. A queued request whose client
+# disconnects releases its place immediately, so a patient deadline costs nothing
+# when nobody is still listening.
+# Limits are PER WORKER PROCESS and, when left at 0, derived from the CPU budget this
+# process actually has (cgroup quota, not the host's core count) divided by
+# HINDSIGHT_API_WORKERS. A positive value overrides the derivation; a NEGATIVE value
+# disables that lane entirely (0 cannot mean "off", because it means "derive").
+# in_flight is a latency target, not a capacity limit: throughput is cores divided by
+# CPU-per-request either way. Too low throttles I/O-bound work, too high rebuilds the
+# queue this exists to prevent.
+# HINDSIGHT_API_ADMISSION_RECALL_MAX_IN_FLIGHT=0
+# HINDSIGHT_API_ADMISSION_RECALL_MAX_WAIT_MS=30000
+# HINDSIGHT_API_ADMISSION_REFLECT_MAX_IN_FLIGHT=0
+# HINDSIGHT_API_ADMISSION_REFLECT_MAX_WAIT_MS=5000
+# HINDSIGHT_API_ADMISSION_RETAIN_MAX_IN_FLIGHT=0
+# HINDSIGHT_API_ADMISSION_RETAIN_MAX_WAIT_MS=2000
+
 # Recall pipeline stages (all on by default). Each is hierarchical, so a single
 # bank can switch a stage off via the config API without changing the server
 # default. Turning all four off reduces recall to a single vector query, the

hindsight-api-slim/hindsight_api/_thread_limits.py

@@ -95,6 +95,17 @@ def _available_cpu_count() -> int:
     return max(1, min(candidates))
 
 
+def available_cpu_count() -> int:
+    """Public alias for :func:`_available_cpu_count`.
+
+    Admission-control defaults size themselves from the real CPU budget, and that
+    budget must come from here rather than ``os.cpu_count()``: under ``--cpus`` the
+    latter reports the host's cores, which would size limits for a machine the
+    process cannot use.
+    """
+    return _available_cpu_count()
+
+
 def default_native_thread_count() -> int:
     """Per-pool cap: ``_MAX_NATIVE_THREADS``, or available CPUs if fewer."""
     return min(_MAX_NATIVE_THREADS, _available_cpu_count())

hindsight-api-slim/hindsight_api/api/admission.py

@@ -0,0 +1,275 @@
+"""Admission control: bound the wait, not just the concurrency.
+
+The engine already limits concurrent work (``recall_max_concurrent`` and friends),
+but a bare ``async with semaphore`` is *backpressure*, not admission control: it
+caps how much runs at once and lets an unbounded queue form behind it. Measured on
+a 2-vCPU container, 1024 concurrent recalls against a 32-permit semaphore produced
+a 12.8s p50 — the latency did not go away, it moved from the event loop into the
+semaphore queue, and the server spent CPU on responses whose clients had long
+since given up.
+
+What is missing is a bound on *waiting*. A lane here has two numbers:
+
+* ``max_in_flight`` — how much of this operation runs concurrently;
+* ``max_wait_seconds`` — how long a request may queue before it is refused.
+
+A request that cannot be admitted within its deadline gets 503 with ``Retry-After``
+immediately, which is a far better answer than a response that arrives after the
+caller timed out. Capacity is unchanged; what changes is that the capacity stops
+being spent on work nobody is waiting for.
+
+**Where this runs.** As a FastAPI dependency on the same routes that carry
+``precheck_for``, so a rejection happens *before the request body is deserialised*
+— the cheapest possible point to say no, and the same place an extension already
+rejects on quota.
+
+**Why a lane per operation rather than one global limit.** Per-request cost spans
+three orders of magnitude on this API (measured: ~0.1ms for ``/health/live``,
+~0.5ms for bank stats, ~23ms for a recall). A single global request cap calibrated
+for recall would throttle health checks that the server can serve 100x faster; one
+calibrated for health would never engage for recall. The operations worth limiting
+are exactly the ones :class:`PrecheckOperation` already enumerates.
+
+**The limits are PER WORKER PROCESS.** Each ``--workers N`` process imports the app
+and builds its own controller, so the process-wide budget is ``N x max_in_flight``.
+Size a lane for one worker, not for the cluster.
+
+A plain ``asyncio.Semaphore`` is correct here because one process runs one event
+loop. (It would not be under ``--event-loops > 1`` on a free-threaded build, where
+each loop would get its own semaphore and admit N times the limit — if multi-loop
+serving ever comes back, this needs ``CrossLoopSemaphore`` instead.) The semaphore
+is constructed before any loop is running, which is fine on 3.10+: it binds lazily
+on first await, not at construction.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+
+from ..cancellation import CancellationToken
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class LaneConfig:
+    """Limits for one operation class."""
+
+    #: Concurrent requests of this operation. 0 disables the lane entirely (the
+    #: resolved value -- see `admission_in_flight_for`, where a *negative* env var
+    #: is what resolves to 0, because 0 there means "derive from the CPU budget").
+    max_in_flight: int
+    #: How long a request may queue for a permit before being refused. 0 means
+    #: "never queue": either a permit is free now or the request is rejected.
+    max_wait_seconds: float
+
+    @property
+    def enabled(self) -> bool:
+        return self.max_in_flight > 0
+
+
+class AdmissionRejected(Exception):
+    """Raised when a request could not be admitted within its deadline."""
+
+    def __init__(self, lane: str, waited_seconds: float, limit: int) -> None:
+        self.lane = lane
+        self.waited_seconds = waited_seconds
+        self.limit = limit
+        super().__init__(
+            f"admission refused for {lane!r} after waiting {waited_seconds * 1000:.0f}ms (limit {limit} concurrent)"
+        )
+
+    @property
+    def retry_after_seconds(self) -> int:
+        """Conservative hint for the client, in whole seconds (``Retry-After``)."""
+        return max(1, round(self.waited_seconds)) if self.waited_seconds else 1
+
+
+class AdmissionAbandoned(Exception):
+    """The client disconnected while its request was queued for a permit.
+
+    Distinct from :class:`AdmissionRejected`: there is nobody left to send a 503 to,
+    so the only useful thing to do is stop working on the request.
+    """
+
+    def __init__(self, lane: str) -> None:
+        self.lane = lane
+        super().__init__(f"client disconnected while queued for {lane!r}")
+
+
+class _ClientGone(Exception):
+    """Internal signal from the acquire race."""
+
+
+async def _acquire_unless_abandoned(
+    semaphore: asyncio.Semaphore, timeout: float, abandoned: CancellationToken | None
+) -> None:
+    """Acquire within ``timeout``, giving up early if the client disconnects.
+
+    Raises :class:`_ClientGone` when the client vanished first, ``TimeoutError`` when
+    the deadline passed.
+    """
+    if abandoned is None:
+        await asyncio.wait_for(semaphore.acquire(), timeout=timeout)
+        return
+
+    # Already gone before it even queued: there is no work worth starting, and
+    # checking here also settles the race below, where a free permit and a cancelled
+    # token would otherwise both be "done" and the permit would win.
+    if abandoned.cancelled:
+        raise _ClientGone()
+
+    acquire = asyncio.ensure_future(semaphore.acquire())
+    gone = asyncio.ensure_future(abandoned.wait())
+    try:
+        done, _ = await asyncio.wait({acquire, gone}, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)
+    finally:
+        gone.cancel()
+
+    if acquire in done:
+        acquire.result()
+        return
+
+    # Not getting a permit. Cancelling a pending acquire is safe -- asyncio's
+    # Semaphore wakes the next waiter when a granted-then-cancelled acquire unwinds
+    # -- but a permit can land in the instant we give up, so if the task completed
+    # anyway hand it straight back rather than leak it for the life of the process.
+    acquire.cancel()
+    try:
+        await acquire
+    except asyncio.CancelledError:
+        pass
+    else:
+        semaphore.release()
+
+    if gone in done:
+        raise _ClientGone()
+    raise TimeoutError()
+
+
+@dataclass
+class LaneStats:
+    """Observable counters for one lane. Cheap to read; useful in an incident."""
+
+    admitted: int = 0
+    rejected: int = 0
+    #: Clients that disconnected while queued -- neither served nor refused.
+    abandoned: int = 0
+    in_flight: int = 0
+    queued: int = 0
+    total_wait_seconds: float = 0.0
+
+    @property
+    def mean_wait_ms(self) -> float:
+        total = self.admitted + self.rejected
+        return (self.total_wait_seconds / total * 1000) if total else 0.0
+
+
+class AdmissionController:
+    """Per-operation concurrency limits with a bounded queue wait."""
+
+    def __init__(self, lanes: dict[str, LaneConfig]) -> None:
+        self._configs = lanes
+        self._semaphores: dict[str, asyncio.Semaphore] = {
+            name: asyncio.Semaphore(cfg.max_in_flight) for name, cfg in lanes.items() if cfg.enabled
+        }
+        self._stats: dict[str, LaneStats] = {name: LaneStats() for name in lanes}
+
+    def lane_config(self, operation: str) -> LaneConfig | None:
+        return self._configs.get(operation)
+
+    def stats(self) -> dict[str, LaneStats]:
+        return self._stats
+
+    @asynccontextmanager
+    async def admit(self, operation: str, abandoned: CancellationToken | None = None) -> AsyncIterator[None]:
+        """Hold a permit for ``operation`` for the duration of the block.
+
+        Falls through with no gating when the operation has no lane or the lane is
+        disabled, so an unconfigured operation behaves exactly as it does today.
+
+        ``abandoned`` is the request's client-disconnect token when one is available
+        (recall and reflect carry one — see :mod:`hindsight_api.api.disconnect`). A
+        queued request whose client has gone away gives up its place immediately
+        instead of holding it for the full deadline. That is what makes a *longer*
+        wait safe: the queue self-cleans, so patience costs nothing when nobody is
+        still listening.
+        """
+        semaphore = self._semaphores.get(operation)
+        if semaphore is None:
+            yield
+            return
+
+        config = self._configs[operation]
+        stats = self._stats[operation]
+        started = time.monotonic()
+        stats.queued += 1
+        # "Never queue" mode still goes through acquire() to keep the permit
+        # accounting honest; the deadline is just short enough to succeed only when a
+        # permit is already free.
+        timeout = 0.001 if config.max_wait_seconds <= 0 else config.max_wait_seconds
+        try:
+            await _acquire_unless_abandoned(semaphore, timeout, abandoned)
+        except _ClientGone:
+            waited = time.monotonic() - started
+            # `queued` is decremented once, in the `finally` below.
+            stats.abandoned += 1
+            # Worth logging on its own: callers giving up while queued is the signal
+            # that the deadline is longer than they are willing to wait.
+            logger.info(
+                "admission abandoned: lane=%s waited=%.3fs limit=%d queued=%d",
+                operation,
+                waited,
+                config.max_in_flight,
+                stats.queued,
+            )
+            raise AdmissionAbandoned(operation) from None
+        except (TimeoutError, asyncio.TimeoutError):
+            waited = time.monotonic() - started
+            stats.rejected += 1
+            stats.total_wait_seconds += waited
+            logger.warning(
+                "admission refused: lane=%s waited=%.3fs limit=%d in_flight=%d queued=%d",
+                operation,
+                waited,
+                config.max_in_flight,
+                stats.in_flight,
+                stats.queued,
+            )
+            raise AdmissionRejected(operation, waited, config.max_in_flight) from None
+        finally:
+            stats.queued -= 1
+
+        waited = time.monotonic() - started
+        stats.admitted += 1
+        stats.total_wait_seconds += waited
+        stats.in_flight += 1
+        try:
+            yield
+        finally:
+            stats.in_flight -= 1
+            semaphore.release()
+
+
+def build_controller_from_config(config) -> AdmissionController:
+    """Build the controller from ``HindsightConfig``.
+
+    Only the three high-volume operations get a lane. `files_retain`,
+    `dry_run_extract` and the mental-model routes are administrative and low-volume:
+    gating them would add knobs nobody tunes without protecting anything that
+    actually saturates a worker. They fall through ungated, exactly as before.
+
+    A lane resolving to 0 in-flight is off (set its env var negative).
+    """
+    return AdmissionController(
+        {
+            "recall": LaneConfig(config.admission_recall_max_in_flight, config.admission_recall_max_wait_seconds),
+            "reflect": LaneConfig(config.admission_reflect_max_in_flight, config.admission_reflect_max_wait_seconds),
+            "retain": LaneConfig(config.admission_retain_max_in_flight, config.admission_retain_max_wait_seconds),
+        }
+    )

hindsight-api-slim/hindsight_api/api/http.py

@@ -26,6 +26,7 @@ from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import JSONResponse, Response
 
 from hindsight_api.api import page_markdown
+from hindsight_api.api.admission import AdmissionAbandoned, AdmissionRejected, build_controller_from_config
 from hindsight_api.api.disconnect import ClientDisconnectCancellationMiddleware, get_scope_cancellation_token
 from hindsight_api.api.observability import HttpObservabilityMiddleware
 from hindsight_api.api.passthrough_headers import collect_passthrough_headers
@@ -4925,6 +4926,10 @@ def create_app(
     # re-discover) and the metrics in a pure-ASGI middleware installed below.
     app.router.route_class = UnknownParamsRoute
 
+    # Per-operation admission control, consulted by the `admit_for` dependency on
+    # the heavy routes. One controller per app so the limits are a process budget.
+    app.state.admission = build_controller_from_config(config)
+
     # Register all routes
     _register_routes(app)
 
@@ -5082,6 +5087,44 @@ def _register_routes(app: FastAPI):
         extra_headers = collect_passthrough_headers(request.headers.raw, get_config().extension_passthrough_headers)
         return RequestContext(api_key=api_key, extra_headers=extra_headers)
 
+    def admit_for(operation: PrecheckOperation):
+        """Build a FastAPI dependency that holds an admission permit for the request.
+
+        Yield-style so the permit is held for the whole request and released once the
+        response has been produced. Declared alongside ``precheck_for`` on the heavy
+        routes: FastAPI resolves dependencies before deserialising the body, so an
+        overloaded server refuses without ever reading the payload.
+
+        Returns 503 with ``Retry-After`` rather than queueing indefinitely — see
+        :mod:`hindsight_api.api.admission` for why the wait, not the concurrency, is
+        the thing worth bounding.
+        """
+
+        async def _admit_dep(request: Request):
+            controller = getattr(app.state, "admission", None)
+            if controller is None:
+                yield
+                return
+            # Recall and reflect carry a disconnect token (see api/disconnect.py). A
+            # queued request whose client has gone gives up its place immediately,
+            # which is what makes a patient deadline affordable.
+            abandoned = get_scope_cancellation_token(request.scope)
+            try:
+                async with controller.admit(str(operation), abandoned=abandoned):
+                    yield
+            except AdmissionAbandoned:
+                # Nobody left to answer. Close the request without spending a slot
+                # or building a response.
+                raise HTTPException(status_code=499, detail="client disconnected while queued") from None
+            except AdmissionRejected as e:
+                raise HTTPException(
+                    status_code=503,
+                    detail=(f"Server is at capacity for '{e.lane}' ({e.limit} concurrent); try again shortly."),
+                    headers={"Retry-After": str(e.retry_after_seconds)},
+                ) from None
+
+        return _admit_dep
+
     def precheck_for(operation: PrecheckOperation):
         """
         Build a FastAPI dependency that runs ``OperationValidator.precheck``.
@@ -5672,6 +5715,7 @@ def _register_routes(app: FastAPI):
         http_request: Request,
         request_context: RequestContext = Depends(get_request_context),
         _precheck: None = Depends(precheck_for(PrecheckOperation.RECALL)),
+        _admit: None = Depends(admit_for(PrecheckOperation.RECALL)),
     ):
         """Run a recall and return results with trace."""
         import time
@@ -5924,6 +5968,7 @@ def _register_routes(app: FastAPI):
         http_request: Request,
         request_context: RequestContext = Depends(get_request_context),
         _precheck: None = Depends(precheck_for(PrecheckOperation.REFLECT)),
+        _admit: None = Depends(admit_for(PrecheckOperation.REFLECT)),
     ):
         metrics = get_metrics_collector()
 
@@ -9020,6 +9065,7 @@ def _register_routes(app: FastAPI):
         request: RetainRequest,
         request_context: RequestContext = Depends(get_request_context),
         _precheck: None = Depends(precheck_for(PrecheckOperation.RETAIN)),
+        _admit: None = Depends(admit_for(PrecheckOperation.RETAIN)),
     ):
         """Retain memories with optional async processing."""
         metrics = get_metrics_collector()

hindsight-api-slim/hindsight_api/config.py

@@ -657,6 +657,17 @@ ENV_ENABLE_DRY_RUN_EXTRACT = "HINDSIGHT_API_ENABLE_DRY_RUN_EXTRACT"
 ENV_DEFAULT_BANK_TEMPLATE = "HINDSIGHT_API_DEFAULT_BANK_TEMPLATE"
 ENV_GRAPH_RETRIEVER = "HINDSIGHT_API_GRAPH_RETRIEVER"
 ENV_RECALL_MAX_CONCURRENT = "HINDSIGHT_API_RECALL_MAX_CONCURRENT"
+
+# Admission control. The engine's *_MAX_CONCURRENT caps limit how much runs at once
+# and let an unbounded queue form behind them; these bound how long a request may
+# WAIT before being refused with 503. A lane's MAX_IN_FLIGHT of 0 derives the limit
+# from the CPU budget; a negative value disables the lane.
+ENV_ADMISSION_RECALL_MAX_IN_FLIGHT = "HINDSIGHT_API_ADMISSION_RECALL_MAX_IN_FLIGHT"
+ENV_ADMISSION_RECALL_MAX_WAIT_MS = "HINDSIGHT_API_ADMISSION_RECALL_MAX_WAIT_MS"
+ENV_ADMISSION_REFLECT_MAX_IN_FLIGHT = "HINDSIGHT_API_ADMISSION_REFLECT_MAX_IN_FLIGHT"
+ENV_ADMISSION_REFLECT_MAX_WAIT_MS = "HINDSIGHT_API_ADMISSION_REFLECT_MAX_WAIT_MS"
+ENV_ADMISSION_RETAIN_MAX_IN_FLIGHT = "HINDSIGHT_API_ADMISSION_RETAIN_MAX_IN_FLIGHT"
+ENV_ADMISSION_RETAIN_MAX_WAIT_MS = "HINDSIGHT_API_ADMISSION_RETAIN_MAX_WAIT_MS"
 ENV_RECALL_CONNECTION_BUDGET = "HINDSIGHT_API_RECALL_CONNECTION_BUDGET"
 ENV_RECALL_MAX_QUERY_TOKENS = "HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS"
 ENV_RECALL_DIAGNOSTIC_PHASES = "HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES"
@@ -1468,6 +1479,59 @@ DEFAULT_ENABLE_BANK_LLM_HEALTH = False
 DEFAULT_DEFAULT_BANK_TEMPLATE: dict | None = None  # BankTemplateManifest dict applied to newly-created banks
 DEFAULT_GRAPH_RETRIEVER = "link_expansion"
 DEFAULT_RECALL_MAX_CONCURRENT = 32  # Max concurrent recall operations per worker
+
+# Admission-control defaults. These are PER WORKER PROCESS: with `--workers N` the
+# process budget is N x the value here.
+#
+# Sized for the reference shape of 2 vCPU / 2 workers.
+#
+# `in_flight` does NOT set capacity -- capacity is cores / cpu-per-request, and a
+# c=1024 sweep measured throughput flat at 40-45 rps whether the limit was 8, 16 or
+# 24 per worker. What it sets is queue depth, and hence the latency of an ADMITTED
+# request: client latency ~= MAX_WAIT_MS + in_flight_total / throughput. Measured at
+# c=1024 (2 workers, untuned): 8/worker -> 1.4s p50, 16 -> 1.8s, 24 -> 2.3s,
+# 32 -> 3.3s.
+#
+# It is bounded on BOTH sides, which is why the smallest value is not the best one:
+#
+#   floor    in_flight >= target_rps * service_time / workers
+#            Too low throttles I/O-bound work. A recall that waits 500ms on a slow
+#            embedding provider can only run in_flight/0.5s per second, so 8 permits
+#            would cap a worker at 16 rps -- well under what its CPU could serve.
+#   ceiling  in_flight <= target_latency * throughput / workers
+#            Too high just rebuilds the unbounded queue this exists to prevent.
+#
+# 16 sits between them for the reference shape: ~1.8s p50 under extreme overload
+# (vs 12.8s with no admission control) while leaving headroom for providers slower
+# than the benchmark's. Raise it if provider latency is high, lower it if latency
+# matters more than peak throughput.
+#
+# It is expressed PER CORE so a bigger machine is not throttled to a small box's
+# queue depth: the reference shape is 2 vCPU / 2 workers, i.e. one core per worker,
+# where this yields the measured 16. `admission_in_flight_for` applies it to the
+# CPU budget this process actually has (cgroup quota, not os.cpu_count()).
+DEFAULT_ADMISSION_RECALL_IN_FLIGHT_PER_CORE = 16
+DEFAULT_ADMISSION_RECALL_MAX_IN_FLIGHT = 0  # 0 = derive from cores; set to override
+# 30s, not 1s. The SDKs are patient (the Python client defaults to a 300s request
+# timeout) so a long queue is observable rather than wasted, and a queued request
+# whose client disconnects releases its place immediately -- so patience costs
+# nothing when nobody is still listening. This absorbs a burst instead of refusing
+# it. Under *sustained* overload it is still bufferbloat: the queue fills to the
+# deadline and the back of it is refused anyway, just later. Lower it if the traffic
+# is persistently over capacity rather than spiky.
+DEFAULT_ADMISSION_RECALL_MAX_WAIT_MS = 30000
+# Reflect is LLM-bound: seconds of wall time, little CPU, and already capped
+# downstream by the per-operation LLM semaphore, so the admission lane only needs to
+# stop an unbounded queue forming in front of it. A shorter wait than recall: a
+# reflect already takes seconds, so queueing another 30s on top rarely helps anyone.
+DEFAULT_ADMISSION_REFLECT_IN_FLIGHT_PER_CORE = 16
+DEFAULT_ADMISSION_REFLECT_MAX_IN_FLIGHT = 0
+DEFAULT_ADMISSION_REFLECT_MAX_WAIT_MS = 5000
+# Synchronous retain runs extraction inline; async retain returns as soon as the
+# operation is queued, so this only bites on the synchronous path.
+DEFAULT_ADMISSION_RETAIN_IN_FLIGHT_PER_CORE = 32
+DEFAULT_ADMISSION_RETAIN_MAX_IN_FLIGHT = 0
+DEFAULT_ADMISSION_RETAIN_MAX_WAIT_MS = 2000
 DEFAULT_RECALL_CONNECTION_BUDGET = 4  # Max concurrent DB connections per recall operation
 DEFAULT_RECALL_MAX_QUERY_TOKENS = 500  # Maximum tokens allowed in recall query
 DEFAULT_RECALL_DIAGNOSTIC_PHASES = True  # Record the subset (diagnostic=true) recall phase metrics
@@ -2680,6 +2744,38 @@ def _parse_default_bank_template(raw: str | None) -> dict | None:
     return parsed
 
 
+def admission_in_flight_for(explicit: int, per_core: int, workers: int) -> int:
+    """Resolve an admission lane's in-flight limit.
+
+    Three cases, because "derive" and "off" both need to be expressible:
+
+    * ``explicit > 0``  -- use it verbatim;
+    * ``explicit == 0`` -- derive from the CPU budget (the default);
+    * ``explicit < 0``  -- disable the lane, so the operation is never gated.
+
+    Otherwise the limit is
+    derived from the CPU budget this process actually has, divided by the number of
+    worker processes sharing it -- the limit is PER WORKER, so a 4-core box running
+    4 workers gets the same per-worker depth as a 1-core box running 1.
+
+    Deriving beats a fixed number because the right depth scales with the machine:
+    queue depth trades latency against the risk of throttling I/O-bound work, and
+    both sides of that trade move with core count. It is deliberately floored at
+    ``per_core`` so a fractional-core deployment still admits enough concurrent work
+    to keep its CPU busy while requests wait on embeddings or an LLM.
+    """
+    if explicit > 0:
+        return explicit
+    if explicit < 0:
+        # Negative is the kill switch. It cannot be 0, because 0 is what "derive"
+        # has to mean for an unset env var.
+        return 0
+    from ._thread_limits import available_cpu_count
+
+    cores_per_worker = available_cpu_count() / max(1, workers)
+    return max(per_core, round(per_core * cores_per_worker))
+
+
 @dataclass
 class HindsightConfig:
     """Configuration container for Hindsight API."""
@@ -3044,6 +3140,12 @@ class HindsightConfig:
     # Recall
     graph_retriever: str
     recall_max_concurrent: int
+    admission_recall_max_in_flight: int
+    admission_recall_max_wait_seconds: float
+    admission_reflect_max_in_flight: int
+    admission_reflect_max_wait_seconds: float
+    admission_retain_max_in_flight: int
+    admission_retain_max_wait_seconds: float
     recall_connection_budget: int
     recall_max_query_tokens: int
     recall_diagnostic_phases: bool
@@ -4462,6 +4564,33 @@ class HindsightConfig:
             # Recall
             graph_retriever=os.getenv(ENV_GRAPH_RETRIEVER, DEFAULT_GRAPH_RETRIEVER),
             recall_max_concurrent=int(os.getenv(ENV_RECALL_MAX_CONCURRENT, str(DEFAULT_RECALL_MAX_CONCURRENT))),
+            admission_recall_max_in_flight=admission_in_flight_for(
+                int(os.getenv(ENV_ADMISSION_RECALL_MAX_IN_FLIGHT, str(DEFAULT_ADMISSION_RECALL_MAX_IN_FLIGHT))),
+                DEFAULT_ADMISSION_RECALL_IN_FLIGHT_PER_CORE,
+                int(os.getenv(ENV_WORKERS, "1")),
+            ),
+            admission_recall_max_wait_seconds=float(
+                os.getenv(ENV_ADMISSION_RECALL_MAX_WAIT_MS, str(DEFAULT_ADMISSION_RECALL_MAX_WAIT_MS))
+            )
+            / 1000.0,
+            admission_reflect_max_in_flight=admission_in_flight_for(
+                int(os.getenv(ENV_ADMISSION_REFLECT_MAX_IN_FLIGHT, str(DEFAULT_ADMISSION_REFLECT_MAX_IN_FLIGHT))),
+                DEFAULT_ADMISSION_REFLECT_IN_FLIGHT_PER_CORE,
+                int(os.getenv(ENV_WORKERS, "1")),
+            ),
+            admission_reflect_max_wait_seconds=float(
+                os.getenv(ENV_ADMISSION_REFLECT_MAX_WAIT_MS, str(DEFAULT_ADMISSION_REFLECT_MAX_WAIT_MS))
+            )
+            / 1000.0,
+            admission_retain_max_in_flight=admission_in_flight_for(
+                int(os.getenv(ENV_ADMISSION_RETAIN_MAX_IN_FLIGHT, str(DEFAULT_ADMISSION_RETAIN_MAX_IN_FLIGHT))),
+                DEFAULT_ADMISSION_RETAIN_IN_FLIGHT_PER_CORE,
+                int(os.getenv(ENV_WORKERS, "1")),
+            ),
+            admission_retain_max_wait_seconds=float(
+                os.getenv(ENV_ADMISSION_RETAIN_MAX_WAIT_MS, str(DEFAULT_ADMISSION_RETAIN_MAX_WAIT_MS))
+            )
+            / 1000.0,
             recall_connection_budget=int(
                 os.getenv(ENV_RECALL_CONNECTION_BUDGET, str(DEFAULT_RECALL_CONNECTION_BUDGET))
             ),

hindsight-api-slim/hindsight_api/main.py

@@ -407,6 +407,10 @@ def main():
         uvicorn_config["reload"] = True
     if args.workers > 1:
         uvicorn_config["workers"] = args.workers
+    # Export the worker count so each child process can size its share of the CPU
+    # budget (admission limits are per worker). uvicorn spawns children that
+    # re-import the app, so the environment is the only channel that reaches them.
+    os.environ[ENV_WORKERS] = str(args.workers)
     if args.forwarded_allow_ips:
         uvicorn_config["forwarded_allow_ips"] = args.forwarded_allow_ips
     if args.ssl_keyfile:

hindsight-api-slim/tests/test_admission_control.py

@@ -0,0 +1,262 @@
+"""Tests for per-operation admission control.
+
+The property under test is the one the engine's existing semaphores do NOT have:
+a request that cannot get a permit within its deadline is *refused*, rather than
+queueing until the caller has given up.
+"""
+
+import asyncio
+
+import pytest
+
+from hindsight_api.api.admission import (
+    AdmissionController,
+    AdmissionRejected,
+    LaneConfig,
+)
+
+
+def _controller(**lanes: LaneConfig) -> AdmissionController:
+    return AdmissionController(dict(lanes))
+
+
+async def test_admits_up_to_the_limit():
+    controller = _controller(recall=LaneConfig(max_in_flight=2, max_wait_seconds=1.0))
+    async with controller.admit("recall"):
+        async with controller.admit("recall"):
+            assert controller.stats()["recall"].in_flight == 2
+    assert controller.stats()["recall"].in_flight == 0
+    assert controller.stats()["recall"].admitted == 2
+
+
+async def test_rejects_when_full_rather_than_queueing_forever():
+    """The whole point: a bounded wait, then 503 -- not an unbounded queue."""
+    controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=0.05))
+
+    async with controller.admit("recall"):
+        started = asyncio.get_running_loop().time()
+        with pytest.raises(AdmissionRejected) as excinfo:
+            async with controller.admit("recall"):
+                pytest.fail("should not have been admitted")
+        waited = asyncio.get_running_loop().time() - started
+
+    # Refused at roughly the deadline, not after the holder finished.
+    assert 0.04 <= waited < 1.0
+    assert excinfo.value.lane == "recall"
+    assert excinfo.value.limit == 1
+    assert excinfo.value.retry_after_seconds >= 1
+    assert controller.stats()["recall"].rejected == 1
+
+
+async def test_permit_is_released_and_a_waiter_then_proceeds():
+    controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=2.0))
+    order: list[str] = []
+
+    async def holder():
+        async with controller.admit("recall"):
+            order.append("holder-in")
+            await asyncio.sleep(0.05)
+        order.append("holder-out")
+
+    async def waiter():
+        await asyncio.sleep(0.01)
+        async with controller.admit("recall"):
+            order.append("waiter-in")
+
+    await asyncio.gather(holder(), waiter())
+    assert order == ["holder-in", "holder-out", "waiter-in"]
+
+
+async def test_permit_released_when_the_body_raises():
+    """A failing request must not leak its permit -- otherwise the lane shrinks."""
+    controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=0.05))
+
+    with pytest.raises(ValueError):
+        async with controller.admit("recall"):
+            raise ValueError("boom")
+
+    assert controller.stats()["recall"].in_flight == 0
+    # The permit came back, so the next request is admitted.
+    async with controller.admit("recall"):
+        pass
+    assert controller.stats()["recall"].admitted == 2
+
+
+async def test_lanes_are_independent():
+    """A saturated lane must not refuse a different operation."""
+    controller = _controller(
+        recall=LaneConfig(max_in_flight=1, max_wait_seconds=0.05),
+        reflect=LaneConfig(max_in_flight=1, max_wait_seconds=0.05),
+    )
+    async with controller.admit("recall"):
+        # reflect has its own permit and is unaffected by recall being full.
+        async with controller.admit("reflect"):
+            pass
+        with pytest.raises(AdmissionRejected):
+            async with controller.admit("recall"):
+                pass
+
+
+async def test_unknown_operation_falls_through_ungated():
+    controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=0.05))
+    async with controller.admit("not_a_lane"):
+        pass  # no exception, no accounting
+
+
+async def test_lane_disabled_by_zero_in_flight():
+    """0 is the documented kill switch and must not gate anything."""
+    controller = _controller(recall=LaneConfig(max_in_flight=0, max_wait_seconds=1.0))
+    async with controller.admit("recall"):
+        async with controller.admit("recall"):
+            pass
+    assert controller.stats()["recall"].admitted == 0
+
+
+async def test_zero_wait_refuses_immediately_when_full():
+    controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=0.0))
+    async with controller.admit("recall"):
+        started = asyncio.get_running_loop().time()
+        with pytest.raises(AdmissionRejected):
+            async with controller.admit("recall"):
+                pass
+        assert asyncio.get_running_loop().time() - started < 0.05
+
+
+class TestDerivedDefaults:
+    """The in-flight default is derived from the CPU budget, not hardcoded."""
+
+    def test_explicit_value_always_wins(self):
+        from hindsight_api.config import admission_in_flight_for
+
+        assert admission_in_flight_for(7, per_core=16, workers=4) == 7
+
+    def test_derives_from_cores_divided_by_workers(self, monkeypatch):
+        from hindsight_api import config as config_mod
+
+        monkeypatch.setattr("hindsight_api._thread_limits.available_cpu_count", lambda: 8)
+        # 8 cores / 2 workers = 4 cores each -> 4 x 16
+        assert config_mod.admission_in_flight_for(0, per_core=16, workers=2) == 64
+
+    def test_floored_at_per_core_on_a_fractional_core_box(self, monkeypatch):
+        """A worker with less than a core still needs enough depth to stay busy
+        while requests wait on embeddings or an LLM."""
+        from hindsight_api import config as config_mod
+
+        monkeypatch.setattr("hindsight_api._thread_limits.available_cpu_count", lambda: 1)
+        assert config_mod.admission_in_flight_for(0, per_core=16, workers=4) == 16
+
+    def test_reference_shape_matches_the_measured_default(self, monkeypatch):
+        """2 vCPU / 2 workers is the shape the 16 was calibrated on."""
+        from hindsight_api import config as config_mod
+
+        monkeypatch.setattr("hindsight_api._thread_limits.available_cpu_count", lambda: 2)
+        assert config_mod.admission_in_flight_for(0, per_core=16, workers=2) == 16
+
+
+class TestAbandonedWhileQueued:
+    """A queued request whose client disconnected must free its place at once.
+
+    This is what makes a patient deadline affordable: without it a 30s wait means
+    30s of held connections for callers that may already be gone.
+    """
+
+    async def test_disconnect_releases_the_waiter_immediately(self):
+        from hindsight_api.api.admission import AdmissionAbandoned
+        from hindsight_api.cancellation import CancellationToken
+
+        controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=30.0))
+        token = CancellationToken()
+
+        async with controller.admit("recall"):
+            started = asyncio.get_running_loop().time()
+
+            async def disconnect_soon():
+                await asyncio.sleep(0.05)
+                token.cancel("client disconnected")
+
+            async def queued():
+                async with controller.admit("recall", abandoned=token):
+                    pytest.fail("should not have been admitted")
+
+            disconnector = asyncio.create_task(disconnect_soon())
+            with pytest.raises(AdmissionAbandoned):
+                await queued()
+            waited = asyncio.get_running_loop().time() - started
+            await disconnector
+
+        # Gave up on disconnect, not after the 30s deadline.
+        assert waited < 1.0
+
+    async def test_abandoned_waiter_does_not_consume_a_permit(self):
+        """The permit must still be there for the next real caller."""
+        from hindsight_api.api.admission import AdmissionAbandoned
+        from hindsight_api.cancellation import CancellationToken
+
+        controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=5.0))
+        token = CancellationToken()
+        token.cancel("gone before it even queued")
+
+        with pytest.raises(AdmissionAbandoned):
+            async with controller.admit("recall", abandoned=token):
+                pass
+
+        # Lane is intact: a live caller is admitted straight away.
+        async with controller.admit("recall"):
+            assert controller.stats()["recall"].in_flight == 1
+        assert controller.stats()["recall"].in_flight == 0
+
+    async def test_live_client_still_admitted_normally(self):
+        """An un-cancelled token must not change the happy path."""
+        from hindsight_api.cancellation import CancellationToken
+
+        controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=1.0))
+        async with controller.admit("recall", abandoned=CancellationToken()):
+            assert controller.stats()["recall"].in_flight == 1
+        assert controller.stats()["recall"].admitted == 1
+
+    async def test_abandoned_waiter_leaves_queue_count_at_zero(self):
+        """The abandon path must decrement `queued` exactly once."""
+        from hindsight_api.api.admission import AdmissionAbandoned
+        from hindsight_api.cancellation import CancellationToken
+
+        controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=5.0))
+        token = CancellationToken()
+        token.cancel("gone")
+
+        with pytest.raises(AdmissionAbandoned):
+            async with controller.admit("recall", abandoned=token):
+                pass
+
+        stats = controller.stats()["recall"]
+        assert stats.queued == 0
+        assert stats.abandoned == 1
+
+    def test_negative_disables_the_lane(self, monkeypatch):
+        """0 means "derive", so the kill switch has to be a negative value."""
+        from hindsight_api import config as config_mod
+
+        monkeypatch.setattr("hindsight_api._thread_limits.available_cpu_count", lambda: 8)
+        assert config_mod.admission_in_flight_for(-1, per_core=16, workers=1) == 0
+
+
+async def test_http_recall_refused_with_503_and_retry_after_when_lane_full(memory):
+    """End to end through the route dependency: a full lane answers 503 + Retry-After."""
+    import httpx
+
+    from hindsight_api.api import create_app
+
+    app = create_app(memory, initialize_memory=False)
+    controller = _controller(recall=LaneConfig(max_in_flight=1, max_wait_seconds=0.0))
+    app.state.admission = controller
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        async with controller.admit("recall"):
+            response = await client.post(
+                "/v1/default/banks/admission-test/memories/recall",
+                json={"query": "anything"},
+            )
+
+    assert response.status_code == 503
+    assert int(response.headers["retry-after"]) >= 1
+    assert controller.stats()["recall"].rejected == 1

hindsight-clients/python/hindsight_client/hindsight_client.py

@@ -7,13 +7,16 @@ easy-to-use interface on top of the auto-generated OpenAPI client.
 
 import asyncio
 import json
+import random
 import warnings
 from datetime import datetime
 from importlib import metadata
 from pathlib import Path
+from collections.abc import Awaitable, Callable
 from typing import Any, Literal
 
 import hindsight_client_api
+from hindsight_client_api.exceptions import ApiException
 
 try:
     _CLIENT_VERSION = metadata.version("hindsight-client")
@@ -125,6 +128,63 @@ def _trigger_input(trigger: dict[str, Any]) -> Any:
     return model(**unnamed_defaults, **trigger)
 
 
+#: Attempts a retryable call makes in total, including the first.
+DEFAULT_MAX_ATTEMPTS = 3
+
+#: Fallback backoff when the server sends 503 without a usable ``Retry-After``.
+_FALLBACK_BACKOFF_SECONDS = 0.5
+
+
+async def _retry_on_capacity(
+    call: "Callable[[], Awaitable[Any]]",
+    max_attempts: int,
+    rng: "random.Random",
+) -> Any:
+    """Run ``call``, retrying while the server reports it is at capacity.
+
+    Only for **idempotent** operations. Recall and reflect are reads, so a repeat is
+    free; synchronous retain is not, and is deliberately excluded — its
+    ``operation_id`` is ignored, so a retry there could duplicate a write.
+
+    Two things matter more than the retry itself:
+
+    ``Retry-After`` is honoured. The server sends it precisely because it knows how
+    long its queue is; retrying sooner just earns another 503.
+
+    The wait is **jittered**. A burst of clients that all receive ``Retry-After: 1``
+    and obey it exactly will come back in lockstep and rebuild the spike that caused
+    the rejection. Spreading them over the interval is what makes retrying safe, and
+    it is the part the generated client's ``ExponentialRetry`` does not do.
+    """
+    for attempt in range(1, max_attempts + 1):
+        try:
+            return await call()
+        except ApiException as e:
+            at_capacity = e.status in (429, 503)
+            if not at_capacity or attempt == max_attempts:
+                raise
+            wait = _retry_after_seconds(e) or _FALLBACK_BACKOFF_SECONDS * (2 ** (attempt - 1))
+            # Full jitter: sleep somewhere in [0, wait], so a synchronised burst
+            # spreads out instead of returning together.
+            await asyncio.sleep(rng.uniform(0, wait))
+    raise AssertionError("unreachable")  # pragma: no cover
+
+
+def _retry_after_seconds(e: "ApiException") -> float | None:
+    """Parse ``Retry-After`` (delta-seconds form) from a response, if present."""
+    headers = getattr(e, "headers", None)
+    if not headers:
+        return None
+    raw = headers.get("Retry-After") or headers.get("retry-after")
+    if raw is None:
+        return None
+    try:
+        return max(0.0, float(raw))
+    except (TypeError, ValueError):
+        # HTTP-date form; the fallback backoff is a better answer than parsing dates.
+        return None
+
+
 class Hindsight:
     """
     High-level, easy-to-use Hindsight API client.
@@ -199,6 +259,7 @@ class Hindsight:
         api_key: str | None = None,
         timeout: float = 300.0,
         user_agent: str | None = None,
+        max_attempts: int = DEFAULT_MAX_ATTEMPTS,
     ):
         """
         Initialize the Hindsight client.
@@ -211,11 +272,19 @@ class Hindsight:
                 should set this to identify themselves (e.g.
                 ``"hindsight-crewai/1.2.0"``). Defaults to
                 ``hindsight-client-python/<version>``.
+            max_attempts: Total attempts for *idempotent* calls (recall, reflect)
+                when the server reports it is at capacity (429/503). 1 disables
+                retrying. Waits honour ``Retry-After`` and are jittered; writes are
+                never retried here.
         """
         config = hindsight_client_api.Configuration(host=base_url, access_token=api_key)
         self._api_client = hindsight_client_api.ApiClient(config)
         self._api_client.user_agent = user_agent or DEFAULT_USER_AGENT
         self._timeout = timeout
+        self._max_attempts = max(1, max_attempts)
+        # Per-client RNG so jitter is injectable in tests and independent of any
+        # seeding the calling application does to the global `random` module.
+        self._retry_rng = random.Random()
         self._base_url = base_url.rstrip("/")
         self._api_key = api_key
         if api_key:
@@ -1244,7 +1313,11 @@ class Hindsight:
             temporal_window=temporal_window_obj,
         )
 
-        return await self._memory_api.recall_memories(bank_id, request_obj, _request_timeout=self._timeout)
+        return await _retry_on_capacity(
+            lambda: self._memory_api.recall_memories(bank_id, request_obj, _request_timeout=self._timeout),
+            self._max_attempts,
+            self._retry_rng,
+        )
 
     async def areflect(
         self,
@@ -1336,7 +1409,11 @@ class Hindsight:
             exclude_mental_model_ids=exclude_mental_model_ids,
         )
 
-        return await self._memory_api.reflect(bank_id, request_obj, _request_timeout=self._timeout)
+        return await _retry_on_capacity(
+            lambda: self._memory_api.reflect(bank_id, request_obj, _request_timeout=self._timeout),
+            self._max_attempts,
+            self._retry_rng,
+        )
 
     # Mental Models methods
 

hindsight-clients/python/tests/test_retry_on_capacity.py

@@ -0,0 +1,127 @@
+"""Retry policy for idempotent calls when the server is at capacity.
+
+The server's admission control answers 503 with ``Retry-After`` when a lane is
+full. Retrying is only safe if it honours that hint and spreads the retries out:
+a burst that all obey ``Retry-After: 1`` exactly comes back in lockstep and
+rebuilds the spike that caused the rejection.
+"""
+
+import random
+
+import pytest
+
+from hindsight_client.hindsight_client import (
+    _retry_after_seconds,
+    _retry_on_capacity,
+)
+from hindsight_client_api.exceptions import ApiException
+
+
+def _at_capacity(status: int = 503, retry_after: str | None = "2") -> ApiException:
+    e = ApiException(status=status)
+    e.headers = {"Retry-After": retry_after} if retry_after is not None else {}
+    return e
+
+
+class TestRetryAfterParsing:
+    def test_reads_delta_seconds(self):
+        assert _retry_after_seconds(_at_capacity(retry_after="7")) == 7.0
+
+    def test_missing_header_returns_none(self):
+        assert _retry_after_seconds(_at_capacity(retry_after=None)) is None
+
+    def test_http_date_form_falls_back(self):
+        """Date form is legal but rarer; the caller's backoff is a better answer."""
+        assert _retry_after_seconds(_at_capacity(retry_after="Wed, 21 Oct 2026 07:28:00 GMT")) is None
+
+    def test_negative_is_clamped(self):
+        assert _retry_after_seconds(_at_capacity(retry_after="-5")) == 0.0
+
+
+class TestRetryBehaviour:
+    async def test_succeeds_without_retrying(self):
+        calls = []
+
+        async def call():
+            calls.append(1)
+            return "ok"
+
+        assert await _retry_on_capacity(call, 3, random.Random(0)) == "ok"
+        assert len(calls) == 1
+
+    async def test_retries_503_then_succeeds(self):
+        calls = []
+
+        async def call():
+            calls.append(1)
+            if len(calls) < 3:
+                raise _at_capacity(503, "0")
+            return "ok"
+
+        assert await _retry_on_capacity(call, 3, random.Random(0)) == "ok"
+        assert len(calls) == 3
+
+    async def test_retries_429_too(self):
+        calls = []
+
+        async def call():
+            calls.append(1)
+            if len(calls) < 2:
+                raise _at_capacity(429, "0")
+            return "ok"
+
+        assert await _retry_on_capacity(call, 3, random.Random(0)) == "ok"
+
+    async def test_gives_up_after_max_attempts_and_reraises(self):
+        calls = []
+
+        async def call():
+            calls.append(1)
+            raise _at_capacity(503, "0")
+
+        with pytest.raises(ApiException) as excinfo:
+            await _retry_on_capacity(call, 3, random.Random(0))
+        assert excinfo.value.status == 503
+        assert len(calls) == 3
+
+    async def test_does_not_retry_other_errors(self):
+        """A 400 is the caller's problem; repeating it just wastes a round trip."""
+        calls = []
+
+        async def call():
+            calls.append(1)
+            raise ApiException(status=400)
+
+        with pytest.raises(ApiException):
+            await _retry_on_capacity(call, 3, random.Random(0))
+        assert len(calls) == 1
+
+    async def test_max_attempts_of_one_disables_retrying(self):
+        calls = []
+
+        async def call():
+            calls.append(1)
+            raise _at_capacity(503, "0")
+
+        with pytest.raises(ApiException):
+            await _retry_on_capacity(call, 1, random.Random(0))
+        assert len(calls) == 1
+
+    async def test_wait_is_jittered_within_retry_after(self, monkeypatch):
+        """Two clients given the same Retry-After must not wake together."""
+        slept: list[float] = []
+
+        async def fake_sleep(seconds):
+            slept.append(seconds)
+
+        monkeypatch.setattr("hindsight_client.hindsight_client.asyncio.sleep", fake_sleep)
+
+        async def call():
+            raise _at_capacity(503, "4")
+
+        for seed in (1, 2, 3):
+            with pytest.raises(ApiException):
+                await _retry_on_capacity(call, 2, random.Random(seed))
+
+        assert all(0 <= s <= 4 for s in slept), slept
+        assert len(set(slept)) > 1, "identical waits: jitter is not being applied"

hindsight-clients/typescript/src/index.ts

@@ -85,6 +85,51 @@ export const CLIENT_VERSION: string =
   typeof __CLIENT_VERSION__ !== "undefined" ? __CLIENT_VERSION__ : "0.0.0-dev";
 export const DEFAULT_USER_AGENT = `hindsight-client-typescript/${CLIENT_VERSION}`;
 
+/** Attempts a retryable call makes in total, including the first. */
+export const DEFAULT_MAX_ATTEMPTS = 3;
+
+/** Fallback backoff when the server sends 503 without a usable `Retry-After`. */
+const FALLBACK_BACKOFF_MS = 500;
+
+/** Parse `Retry-After` (delta-seconds form) into milliseconds, if present. */
+export function retryAfterMs(response: Response | undefined): number | null {
+  const raw = response?.headers?.get("retry-after");
+  if (raw === null || raw === undefined) return null;
+  const seconds = Number(raw);
+  // The HTTP-date form is legal but rare; the caller's backoff beats parsing dates.
+  if (!Number.isFinite(seconds)) return null;
+  return Math.max(0, seconds * 1000);
+}
+
+/**
+ * Run `send`, retrying while the server reports it is at capacity (429/503).
+ *
+ * Only for **idempotent** operations. Recall and reflect are reads, so a repeat is
+ * free; synchronous retain is not, and is deliberately excluded — its
+ * `operation_id` is ignored there, so a retry could duplicate a write.
+ *
+ * Two things matter more than the retry itself. `Retry-After` is honoured, because
+ * the server sends it knowing how long its own queue is. And the wait is
+ * **jittered**: a burst of clients that all receive `Retry-After: 1` and obey it
+ * exactly return in lockstep and rebuild the spike that caused the rejection.
+ */
+export async function retryOnCapacity<T extends { response?: Response }>(
+  send: () => Promise<T>,
+  maxAttempts: number,
+  random: () => number = Math.random
+): Promise<T> {
+  let result = await send();
+  for (let attempt = 1; attempt < maxAttempts; attempt++) {
+    const status = result.response?.status;
+    if (status !== 429 && status !== 503) return result;
+    const wait = retryAfterMs(result.response) ?? FALLBACK_BACKOFF_MS * 2 ** (attempt - 1);
+    // Full jitter: sleep somewhere in [0, wait] so a synchronised burst spreads out.
+    await new Promise((resolve) => setTimeout(resolve, random() * wait));
+    result = await send();
+  }
+  return result;
+}
+
 export interface HindsightClientOptions {
   baseUrl: string;
   /**
@@ -100,6 +145,12 @@ export interface HindsightClientOptions {
   userAgent?: string;
   /** Optional headers sent with every request. */
   headers?: Record<string, string>;
+  /**
+   * Total attempts for *idempotent* calls (recall, reflect) when the server reports
+   * it is at capacity (429/503). 1 disables retrying. Waits honour `Retry-After`
+   * and are jittered; writes are never retried.
+   */
+  maxAttempts?: number;
 }
 
 /**
@@ -236,6 +287,7 @@ function warnIfOperationIdDropped(
 
 export class HindsightClient {
   private client: Client;
+  private maxAttempts: number;
 
   constructor(options: HindsightClientOptions) {
     const headers: Record<string, string> = {
@@ -251,6 +303,7 @@ export class HindsightClient {
         headers,
       })
     );
+    this.maxAttempts = Math.max(1, options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS);
   }
 
   /**
@@ -478,39 +531,43 @@ export class HindsightClient {
       signal?: AbortSignal;
     }
   ): Promise<RecallResponse> {
-    const response = await sdk.recallMemories({
-      client: this.client,
-      path: { bank_id: bankId },
-      body: {
-        query,
-        types: options?.types,
-        prefer_observations: options?.preferObservations,
-        max_tokens: options?.maxTokens,
-        budget: options?.budget || "mid",
-        trace: options?.trace,
-        query_timestamp: options?.queryTimestamp,
-        include: {
-          entities:
-            options?.includeEntities === false
-              ? null
-              : options?.includeEntities
-                ? { max_tokens: options?.maxEntityTokens ?? 500 }
+    const response = await retryOnCapacity(
+      () =>
+        sdk.recallMemories({
+          client: this.client,
+          path: { bank_id: bankId },
+          body: {
+            query,
+            types: options?.types,
+            prefer_observations: options?.preferObservations,
+            max_tokens: options?.maxTokens,
+            budget: options?.budget || "mid",
+            trace: options?.trace,
+            query_timestamp: options?.queryTimestamp,
+            include: {
+              entities:
+                options?.includeEntities === false
+                  ? null
+                  : options?.includeEntities
+                    ? { max_tokens: options?.maxEntityTokens ?? 500 }
+                    : undefined,
+              chunks: options?.includeChunks
+                ? { max_tokens: options?.maxChunkTokens ?? 8192 }
                 : undefined,
-          chunks: options?.includeChunks
-            ? { max_tokens: options?.maxChunkTokens ?? 8192 }
-            : undefined,
-          source_facts: options?.includeSourceFacts
-            ? { max_tokens: options?.maxSourceFactsTokens ?? 4096 }
-            : undefined,
-        },
-        tags: options?.tags,
-        tags_match: options?.tagsMatch,
-        tag_groups: options?.tagGroups,
-        min_scores: options?.minScores,
-        temporal_window: options?.temporalWindow,
-      },
-      signal: options?.signal,
-    });
+              source_facts: options?.includeSourceFacts
+                ? { max_tokens: options?.maxSourceFactsTokens ?? 4096 }
+                : undefined,
+            },
+            tags: options?.tags,
+            tags_match: options?.tagsMatch,
+            tag_groups: options?.tagGroups,
+            min_scores: options?.minScores,
+            temporal_window: options?.temporalWindow,
+          },
+          signal: options?.signal,
+        }),
+      this.maxAttempts
+    );
 
     return this.validateResponse(response, "recall");
   }
@@ -558,25 +615,29 @@ export class HindsightClient {
               : undefined,
           }
         : undefined;
-    const response = await sdk.reflect({
-      client: this.client,
-      path: { bank_id: bankId },
-      body: {
-        query,
-        context: options?.context,
-        budget: options?.budget || "low",
-        tags: options?.tags,
-        tags_match: options?.tagsMatch,
-        tag_groups: options?.tagGroups,
-        apply_all_directives: options?.applyAllDirectives,
-        response_schema: options?.responseSchema,
-        fact_types: options?.factTypes,
-        exclude_mental_models: options?.excludeMentalModels,
-        exclude_mental_model_ids: options?.excludeMentalModelIds,
-        include,
-      },
-      signal: options?.signal,
-    });
+    const response = await retryOnCapacity(
+      () =>
+        sdk.reflect({
+          client: this.client,
+          path: { bank_id: bankId },
+          body: {
+            query,
+            context: options?.context,
+            budget: options?.budget || "low",
+            tags: options?.tags,
+            tags_match: options?.tagsMatch,
+            tag_groups: options?.tagGroups,
+            apply_all_directives: options?.applyAllDirectives,
+            response_schema: options?.responseSchema,
+            fact_types: options?.factTypes,
+            exclude_mental_models: options?.excludeMentalModels,
+            exclude_mental_model_ids: options?.excludeMentalModelIds,
+            include,
+          },
+          signal: options?.signal,
+        }),
+      this.maxAttempts
+    );
 
     return this.validateResponse(response, "reflect");
   }

hindsight-clients/typescript/tests/retry.test.ts

@@ -0,0 +1,119 @@
+/**
+ * Retry policy for idempotent calls when the server is at capacity.
+ *
+ * Mirrors tests/test_retry_on_capacity.py in the Python wrapper: the two
+ * maintained wrappers are expected to expose the same behaviour.
+ */
+import { retryAfterMs, retryOnCapacity } from "../src/index";
+
+function res(status: number, retryAfter?: string): { response: Response } {
+  const headers = new Headers();
+  if (retryAfter !== undefined) headers.set("retry-after", retryAfter);
+  return { response: { status, headers } as unknown as Response };
+}
+
+describe("retryAfterMs", () => {
+  it("reads delta-seconds", () => {
+    expect(retryAfterMs(res(503, "7").response)).toBe(7000);
+  });
+
+  it("returns null when absent", () => {
+    expect(retryAfterMs(res(503).response)).toBeNull();
+  });
+
+  it("falls back on the HTTP-date form", () => {
+    expect(retryAfterMs(res(503, "Wed, 21 Oct 2026 07:28:00 GMT").response)).toBeNull();
+  });
+
+  it("clamps a negative value", () => {
+    expect(retryAfterMs(res(503, "-5").response)).toBe(0);
+  });
+});
+
+describe("retryOnCapacity", () => {
+  it("does not retry a success", async () => {
+    let calls = 0;
+    const send = async () => {
+      calls++;
+      return res(200);
+    };
+    await retryOnCapacity(send, 3, () => 0);
+    expect(calls).toBe(1);
+  });
+
+  it("retries a 503 and returns the eventual success", async () => {
+    let calls = 0;
+    const send = async () => {
+      calls++;
+      return calls < 3 ? res(503, "0") : res(200);
+    };
+    const out = await retryOnCapacity(send, 3, () => 0);
+    expect(calls).toBe(3);
+    expect(out.response?.status).toBe(200);
+  });
+
+  it("retries a 429 as well", async () => {
+    let calls = 0;
+    const send = async () => {
+      calls++;
+      return calls < 2 ? res(429, "0") : res(200);
+    };
+    await retryOnCapacity(send, 3, () => 0);
+    expect(calls).toBe(2);
+  });
+
+  it("gives up after maxAttempts and returns the last response", async () => {
+    let calls = 0;
+    const send = async () => {
+      calls++;
+      return res(503, "0");
+    };
+    const out = await retryOnCapacity(send, 3, () => 0);
+    expect(calls).toBe(3);
+    expect(out.response?.status).toBe(503);
+  });
+
+  it("does not retry other failures", async () => {
+    let calls = 0;
+    const send = async () => {
+      calls++;
+      return res(400);
+    };
+    await retryOnCapacity(send, 3, () => 0);
+    expect(calls).toBe(1);
+  });
+
+  it("maxAttempts of 1 disables retrying", async () => {
+    let calls = 0;
+    const send = async () => {
+      calls++;
+      return res(503, "0");
+    };
+    await retryOnCapacity(send, 1, () => 0);
+    expect(calls).toBe(1);
+  });
+
+  it("jitters the wait within Retry-After", async () => {
+    // Two clients handed the same Retry-After must not wake together.
+    const waits: number[] = [];
+    const realSetTimeout = global.setTimeout;
+    // @ts-expect-error test double
+    global.setTimeout = (fn: () => void, ms: number) => {
+      waits.push(ms);
+      return realSetTimeout(fn, 0);
+    };
+    try {
+      for (const r of [0.1, 0.5, 0.9]) {
+        await retryOnCapacity(
+          async () => res(503, "4"),
+          2,
+          () => r
+        );
+      }
+    } finally {
+      global.setTimeout = realSetTimeout;
+    }
+    expect(waits.every((w) => w >= 0 && w <= 4000)).toBe(true);
+    expect(new Set(waits).size).toBeGreaterThan(1);
+  });
+});

hindsight-docs/docs/developer/configuration.md

@@ -1420,6 +1420,12 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust
 | `HINDSIGHT_API_LINK_EXPANSION_TIMEOUT` | Timeout (seconds) for the per-entity graph expansion query in `link_expansion` retrieval. | `10` |
 | `HINDSIGHT_API_RECALL_MAX_CONCURRENT` | Max concurrent recall operations per worker (backpressure) | `32` |
 | `HINDSIGHT_API_RECALL_CONNECTION_BUDGET` | Max concurrent DB connections per recall operation | `4` |
+| `HINDSIGHT_API_ADMISSION_RECALL_MAX_IN_FLIGHT` | Concurrent recalls admitted per worker before requests queue. `0` derives it from the CPU budget this process has (cgroup quota) divided by `HINDSIGHT_API_WORKERS`; a negative value disables the lane. A latency target, not a capacity limit: throughput is unchanged either way, but too low throttles I/O-bound work and too high rebuilds the queue. | `0` (derived) |
+| `HINDSIGHT_API_ADMISSION_RECALL_MAX_WAIT_MS` | How long a recall may queue for a slot before being refused with 503 + `Retry-After`. A queued request whose client disconnects releases its place immediately. Absorbs bursts; lower it if load is persistently over capacity, where a long queue just delays the same refusals. | `30000` |
+| `HINDSIGHT_API_ADMISSION_REFLECT_MAX_IN_FLIGHT` | As above, for reflect. | `0` (derived) |
+| `HINDSIGHT_API_ADMISSION_REFLECT_MAX_WAIT_MS` | As above, for reflect. | `5000` |
+| `HINDSIGHT_API_ADMISSION_RETAIN_MAX_IN_FLIGHT` | As above, for retain. Only bites on the synchronous path; an async retain returns as soon as the operation is queued. | `0` (derived) |
+| `HINDSIGHT_API_ADMISSION_RETAIN_MAX_WAIT_MS` | As above, for retain. | `2000` |
 | `HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS` | Maximum token length of a recall query. API requests exceeding this limit are rejected with HTTP 400; recalls that Hindsight runs internally (consolidation, reflect, MCP) truncate the query to the limit instead of failing. `0` disables the limit. | `500` |
 | `HINDSIGHT_API_QUERY_ANALYZER_LANGUAGES` | Restrict the locales `dateparser` considers when extracting temporal constraints from a recall query, as a comma-separated list of language codes (e.g. `en` or `en,zh`). Empty keeps full auto-detection across all supported locales. Restricting is significantly faster (auto-detection dominates recall's CPU cost) and avoids locale misdetection on a known-language corpus, but explicit dates written in an unlisted locale will then misparse rather than yield no constraint — only set this when you know which languages your queries use. Does not affect Chinese, which is handled before `dateparser` runs. | _(empty)_ |
 | `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` | Max candidates to rerank per recall (RRF pre-filters the rest) | `300` |

hindsight-embed/hindsight_embed/env.example

@@ -196,6 +196,10 @@ HINDSIGHT_API_LOG_LEVEL=info
 # HINDSIGHT_API_GZIP_MIN_SIZE=1024
 # Diagnostic: log event-loop lag percentiles every N seconds. 0 disables.
 # HINDSIGHT_API_LOOP_LAG_REPORT_SECONDS=0
+# Record every event-loop lag sample as a histogram (hindsight_event_loop_lag_seconds).
+# HINDSIGHT_API_LOOP_LAG_METRIC=false
+# With --workers N, make /metrics cover every worker, each series labelled api_worker=<slot>.
+# HINDSIGHT_API_METRICS_WORKER_LABEL=false
 # Record the diagnostic (subset) recall phase metrics.
 # HINDSIGHT_API_RECALL_DIAGNOSTIC_PHASES=true
 # Record 1 in N recall-phase metric observations (1 = all). Cuts metrics CPU at high recall rates.
@@ -460,6 +464,25 @@ HINDSIGHT_API_LOG_LEVEL=info
 # Explicit refreshes always run immediately.
 # HINDSIGHT_API_MENTAL_MODEL_MIN_REFRESH_INTERVAL_SECONDS=0
 
+# Admission control. The *_MAX_CONCURRENT caps above bound how much runs at once and
+# let an unbounded queue form behind them; these bound how long a request may WAIT
+# before it is refused with 503 + Retry-After. A queued request whose client
+# disconnects releases its place immediately, so a patient deadline costs nothing
+# when nobody is still listening.
+# Limits are PER WORKER PROCESS and, when left at 0, derived from the CPU budget this
+# process actually has (cgroup quota, not the host's core count) divided by
+# HINDSIGHT_API_WORKERS. A positive value overrides the derivation; a NEGATIVE value
+# disables that lane entirely (0 cannot mean "off", because it means "derive").
+# in_flight is a latency target, not a capacity limit: throughput is cores divided by
+# CPU-per-request either way. Too low throttles I/O-bound work, too high rebuilds the
+# queue this exists to prevent.
+# HINDSIGHT_API_ADMISSION_RECALL_MAX_IN_FLIGHT=0
+# HINDSIGHT_API_ADMISSION_RECALL_MAX_WAIT_MS=30000
+# HINDSIGHT_API_ADMISSION_REFLECT_MAX_IN_FLIGHT=0
+# HINDSIGHT_API_ADMISSION_REFLECT_MAX_WAIT_MS=5000
+# HINDSIGHT_API_ADMISSION_RETAIN_MAX_IN_FLIGHT=0
+# HINDSIGHT_API_ADMISSION_RETAIN_MAX_WAIT_MS=2000
+
 # Recall pipeline stages (all on by default). Each is hierarchical, so a single
 # bank can switch a stage off via the config API without changing the server
 # default. Turning all four off reduces recall to a single vector query, the

skills/hindsight-docs/references/developer/configuration.md

@@ -1420,6 +1420,12 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust
 | `HINDSIGHT_API_LINK_EXPANSION_TIMEOUT` | Timeout (seconds) for the per-entity graph expansion query in `link_expansion` retrieval. | `10` |
 | `HINDSIGHT_API_RECALL_MAX_CONCURRENT` | Max concurrent recall operations per worker (backpressure) | `32` |
 | `HINDSIGHT_API_RECALL_CONNECTION_BUDGET` | Max concurrent DB connections per recall operation | `4` |
+| `HINDSIGHT_API_ADMISSION_RECALL_MAX_IN_FLIGHT` | Concurrent recalls admitted per worker before requests queue. `0` derives it from the CPU budget this process has (cgroup quota) divided by `HINDSIGHT_API_WORKERS`; a negative value disables the lane. A latency target, not a capacity limit: throughput is unchanged either way, but too low throttles I/O-bound work and too high rebuilds the queue. | `0` (derived) |
+| `HINDSIGHT_API_ADMISSION_RECALL_MAX_WAIT_MS` | How long a recall may queue for a slot before being refused with 503 + `Retry-After`. A queued request whose client disconnects releases its place immediately. Absorbs bursts; lower it if load is persistently over capacity, where a long queue just delays the same refusals. | `30000` |
+| `HINDSIGHT_API_ADMISSION_REFLECT_MAX_IN_FLIGHT` | As above, for reflect. | `0` (derived) |
+| `HINDSIGHT_API_ADMISSION_REFLECT_MAX_WAIT_MS` | As above, for reflect. | `5000` |
+| `HINDSIGHT_API_ADMISSION_RETAIN_MAX_IN_FLIGHT` | As above, for retain. Only bites on the synchronous path; an async retain returns as soon as the operation is queued. | `0` (derived) |
+| `HINDSIGHT_API_ADMISSION_RETAIN_MAX_WAIT_MS` | As above, for retain. | `2000` |
 | `HINDSIGHT_API_RECALL_MAX_QUERY_TOKENS` | Maximum token length of a recall query. API requests exceeding this limit are rejected with HTTP 400; recalls that Hindsight runs internally (consolidation, reflect, MCP) truncate the query to the limit instead of failing. `0` disables the limit. | `500` |
 | `HINDSIGHT_API_QUERY_ANALYZER_LANGUAGES` | Restrict the locales `dateparser` considers when extracting temporal constraints from a recall query, as a comma-separated list of language codes (e.g. `en` or `en,zh`). Empty keeps full auto-detection across all supported locales. Restricting is significantly faster (auto-detection dominates recall's CPU cost) and avoids locale misdetection on a known-language corpus, but explicit dates written in an unlisted locale will then misparse rather than yield no constraint — only set this when you know which languages your queries use. Does not affect Chinese, which is handled before `dateparser` runs. | _(empty)_ |
 | `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` | Max candidates to rerank per recall (RRF pre-filters the rest) | `300` |