Capacity recovery: typed errors, shared deadline retry, guard fix, turn checkpoint/resume

docs/configuration.md

@@ -275,3 +275,45 @@ shutdown completes. Recovery therefore does not depend on the service
 unit's `Restart=` policy, Docker restart policy, or any supervisor at all.
 `Restart=always` (what the packaged unit ships) is still recommended so the
 service also recovers from crashes and reboots.
+
+## LLM Recovery (capacity outages)
+
+Model-capacity errors (e.g. `server_is_overloaded`, which arrives inside an
+HTTP 200 as an SSE error event) are retried with a deadline-based policy
+shared by chat, agents, and autonomous loops, coordinated by a per-model
+circuit breaker. Quota handling is unchanged: HTTP 429 still rotates
+accounts inside the provider client; capacity never does.
+
+```yaml
+llm_recovery:
+  generation_deadline_seconds: 300   # retry budget per LLM generation (waiting, not the attempt)
+  backoff_cap_seconds: 45            # full-jitter backoff ceiling between attempts
+  breaker_generation_threshold: 1    # failed generations before the model breaker opens
+  breaker_cooldown_base_seconds: 30  # first cooldown; doubles per failed probe
+  breaker_cooldown_cap_seconds: 300  # cooldown ceiling
+```
+
+All keys are optional (schema defaults shown); the section does not need to
+exist in `config.yml`.
+
+## Turn State (checkpoints and resume)
+
+Discord chat turns are checkpointed to a durable store so a capacity outage
+suspends the turn with its work preserved instead of discarding it. A
+suspended turn auto-resumes when capacity returns (if nothing else has
+happened in the channel), or the original requester can reply `resume`
+within the resumable window. Interrupted tool executions are recorded as
+outcome-unknown and are never re-run automatically.
+
+```yaml
+turn_state:
+  enabled: true
+  db_path: "./data/turn_state/turns.sqlite3"
+  auto_resume: true
+  resume_ttl_hours: 24        # resumable window from last real progress
+  payload_retention_days: 7   # diagnostic payloads, then compacted to tombstones
+  ledger_retention_days: 90   # side-effect ledger (outcome-unknown rows never expire)
+```
+
+All keys are optional; disabling `turn_state.enabled` restores the previous
+behavior (capacity exhaustion ends the turn with an error).

src/agents/manager.py

@@ -31,7 +31,10 @@
 WAIT_POLL_INTERVAL = 2           # poll interval for wait_for_agents
 ITERATION_CB_TIMEOUT = 120       # 2 min timeout per LLM call
 TOOL_EXEC_TIMEOUT = 300          # 5 min timeout per tool execution
-MAX_RECOVERY_ATTEMPTS = 1        # retries before transitioning to FAILED
+# (The manager-level MAX_RECOVERY_ATTEMPTS retry ladder was removed
+# 2026-07-30: transient-failure recovery now lives inside the iteration
+# callback via src/llm/recovery.py. AgentInfo.recovery_attempts remains for
+# API/trajectory shape compatibility and stays 0.)
 MAX_NESTING_DEPTH = 2            # default max sub-agent depth (root=0)
 MAX_CHILDREN_PER_AGENT = 3       # max direct children one agent can spawn
 
@@ -922,7 +925,6 @@ def _check_lifetime() -> bool:
             agent.transition(AgentState.EXECUTING, f"iteration {iteration + 1}")
             agent.last_activity = time.time()
             agent.iteration_count = iteration + 1
-            agent.recovery_attempts = 0  # per-iteration recovery budget
             iter_start = time.time()
 
             # Call LLM with recovery support
@@ -1118,15 +1120,16 @@ async def _call_llm_with_recovery(
     system_prompt: str,
     tools: list[dict],
 ) -> dict | None:
-    """Call LLM with single-retry recovery on transient errors.
+    """Call the LLM for one agent iteration.
 
-    Each wait is bounded by the agent's snapshotted iteration_timeout, capped
-    at the remaining lifetime — the hard deadline must hold even during a
-    long LLM await (the between-iteration lifetime check alone would let a
-    quiet agent overrun it).
+    Transient-failure recovery (capacity/transport/breaker waits) lives
+    INSIDE the iteration callback via the shared deadline-based policy
+    (``src/llm/recovery.py``) — the old manager-level bare-``except`` single
+    retry ladder retried programming defects and is deliberately gone
+    (design settled with Odin, 2026-07-30). What remains here is the wall:
+    the agent's snapshotted iteration_timeout capped at remaining lifetime
+    hard-bounds the callback INCLUDING any recovery waits.
 
-    On first failure: EXECUTING → RECOVERING → EXECUTING (retry).
-    On second failure: EXECUTING → FAILED.
     Returns the LLM response dict, or None if agent reached terminal state.
     """
     remaining = _remaining_lifetime(agent)
@@ -1139,71 +1142,36 @@ async def _call_llm_with_recovery(
             iteration_callback(agent.messages, system_prompt, tools),
             timeout=call_timeout,
         )
-    except (TimeoutError, Exception) as first_err:
-        is_timeout = isinstance(first_err, asyncio.TimeoutError)
-        if is_timeout and _remaining_lifetime(agent) <= 0:
+    except TimeoutError:
+        if _remaining_lifetime(agent) <= 0:
             # The wait was lifetime-capped and the deadline has passed:
             # this is lifetime exhaustion, not a stuck LLM call.
             _lifetime_timeout(agent)
             return None
-        err_desc = (f"LLM timeout after {int(call_timeout)}s" if is_timeout
-                    else f"LLM error: {first_err}")
-
-        if agent.recovery_attempts < MAX_RECOVERY_ATTEMPTS:
-            agent.recovery_attempts += 1
-            agent.transition(AgentState.RECOVERING, err_desc)
-            log.warning(
-                "Agent %s recovering (attempt %d): %s",
-                agent.id,
-                agent.recovery_attempts,
-                err_desc,
-            )
-
-            retry_delay = 1
-            if hasattr(first_err, "retry_after"):
-                retry_delay = min(first_err.retry_after, 90.0)
-                log.info("Agent %s: circuit breaker wait %.0fs", agent.id, retry_delay)
-            # The recovery sleep must not outlive the deadline either — a
-            # 90s breaker wait with 5s of lifetime left sleeps 5s, and the
-            # retry-entry check below then settles it.
-            remaining = _remaining_lifetime(agent)
-            if remaining <= 0:
-                _lifetime_timeout(agent)
-                return None
-            await asyncio.sleep(min(retry_delay, remaining))
-
-            agent.transition(AgentState.EXECUTING, "retry after recovery")
-
-            remaining = _remaining_lifetime(agent)
-            if remaining <= 0:
-                _lifetime_timeout(agent)
-                return None
-            retry_timeout = min(agent.iteration_timeout, remaining)
-            try:
-                return await asyncio.wait_for(
-                    iteration_callback(agent.messages, system_prompt, tools),
-                    timeout=retry_timeout,
-                )
-            except (TimeoutError, Exception) as retry_err:
-                retry_is_timeout = isinstance(retry_err, asyncio.TimeoutError)
-                if retry_is_timeout and _remaining_lifetime(agent) <= 0:
-                    _lifetime_timeout(agent)
-                    return None
-                # str(asyncio.TimeoutError()) is EMPTY — always store the
-                # formatted description, never the bare exception string.
-                retry_desc = (f"retry timed out after {int(retry_timeout)}s"
-                              if retry_is_timeout else f"retry failed: {retry_err}")
-                log.error("Agent %s recovery failed: %s", agent.id, retry_desc)
-                agent.transition(AgentState.FAILED, retry_desc)
-                agent.error = retry_desc
-                agent.ended_at = time.time()
-                return None
-        else:
-            log.error("Agent %s LLM call failed (no retries left): %s", agent.id, err_desc)
-            agent.transition(AgentState.FAILED, err_desc)
-            agent.error = err_desc
-            agent.ended_at = time.time()
+        # str(asyncio.TimeoutError()) is EMPTY — always store the formatted
+        # description, never the bare exception string.
+        err_desc = f"LLM timeout after {int(call_timeout)}s"
+        log.error("Agent %s LLM call failed: %s", agent.id, err_desc)
+        agent.transition(AgentState.FAILED, err_desc)
+        agent.error = err_desc
+        agent.ended_at = time.time()
+        return None
+    except Exception as exc:
+        if _remaining_lifetime(agent) <= 0:
+            # Lifetime exhaustion wins over failure classification (the
+            # v3.59.0 rule: exhaustion is TIMEOUT, never FAILED).
+            _lifetime_timeout(agent)
             return None
+        # Typed fast-fail (auth / malformed request / quota-exhausted after
+        # rotation) or a programming defect: neither earns a manager-level
+        # retry — transient classes were already retried inside the callback
+        # for up to the generation deadline.
+        err_desc = f"LLM error: {exc}" if str(exc) else f"LLM error: {type(exc).__name__}"
+        log.error("Agent %s LLM call failed (no retry): %s", agent.id, err_desc)
+        agent.transition(AgentState.FAILED, err_desc)
+        agent.error = err_desc
+        agent.ended_at = time.time()
+        return None
 
 
 def _get_last_progress(agent: AgentInfo) -> str:

src/config/schema.py

@@ -581,6 +581,34 @@ class GracefulDegradationConfig(BaseModel):
     unavailable_threshold: int = 10  # consecutive failures before UNAVAILABLE
 
 
+class LLMRecoveryConfig(BaseModel):
+    """Deadline-based recovery for logical LLM generations (all three call
+    paths: chat, agents, autonomous loops) plus the model-scoped capacity
+    breaker. The deadline bounds WAITING between attempts, never the
+    attempt itself; capacity never rotates accounts (429 rotation is the
+    provider client's job and is untouched)."""
+
+    generation_deadline_seconds: float = Field(default=300.0, ge=10.0, le=3600.0)
+    backoff_cap_seconds: float = Field(default=45.0, ge=1.0, le=300.0)
+    breaker_generation_threshold: int = Field(default=1, ge=1, le=10)
+    breaker_cooldown_base_seconds: float = Field(default=30.0, ge=1.0, le=600.0)
+    breaker_cooldown_cap_seconds: float = Field(default=300.0, ge=30.0, le=3600.0)
+
+
+class TurnStateConfig(BaseModel):
+    """Durable chat-turn checkpoints, side-effect ledger, and resume.
+
+    Discord chat turns only (v1). Disabled => turns run exactly as before
+    (capacity exhaustion discards work instead of suspending)."""
+
+    enabled: bool = True
+    db_path: str = "./data/turn_state/turns.sqlite3"
+    auto_resume: bool = True
+    resume_ttl_hours: float = Field(default=24.0, ge=1.0, le=24.0 * 14)
+    payload_retention_days: float = Field(default=7.0, ge=1.0, le=90.0)
+    ledger_retention_days: float = Field(default=90.0, ge=30.0, le=365.0)
+
+
 class AuditConfig(BaseModel):
     hmac_key: str = ""  # Empty = signing disabled
 
@@ -886,6 +914,8 @@ class Config(BaseModel):
     grafana_alerts: GrafanaAlertConfig = GrafanaAlertConfig()
     outbound_webhooks: OutboundWebhooksConfig = OutboundWebhooksConfig()
     graceful_degradation: GracefulDegradationConfig = GracefulDegradationConfig()
+    llm_recovery: LLMRecoveryConfig = LLMRecoveryConfig()
+    turn_state: TurnStateConfig = TurnStateConfig()
 
 
 def _substitute_env_vars(text: str) -> str:

src/discord/housekeeping.py

@@ -32,6 +32,7 @@ def __init__(
         loop_agent_bridge,
         channel_logger,
         fts_index,
+        turn_store=None,
     ) -> None:
         self._get_config = get_config
         self._sessions = sessions
@@ -42,6 +43,7 @@ def __init__(
         self._loop_agent_bridge = loop_agent_bridge
         self._channel_logger = channel_logger
         self._fts_index = fts_index
+        self._turn_store = turn_store
 
     def cleanup_stale(self) -> None:
         """Remove stale entries from per-channel caches to prevent memory leaks.
@@ -94,6 +96,33 @@ def cleanup_stale(self) -> None:
             except Exception:
                 pass
 
+        # Turn-state retention: the three clocks (resumable TTL, diagnostic
+        # payload compaction, ledger expiry — OUTCOME_UNKNOWN never expires)
+        # plus the expired-ACTIVE defense sweep (a dead owner's turn must
+        # become visible to the resume path even if the boot sweep missed it).
+        if self._turn_store is not None:
+            try:
+                swept_active = self._turn_store.sweep_expired_active_sync()
+                if any(swept_active.values()):
+                    log.warning(
+                        "Expired-active turn sweep: %s (dead-owner turns "
+                        "suspended)", swept_active,
+                    )
+            except Exception:
+                pass
+            try:
+                config = self._get_config()
+                ts = getattr(config, "turn_state", None)
+                swept = self._turn_store.ttl_sweep_sync(
+                    resume_ttl_hours=getattr(ts, "resume_ttl_hours", 24.0),
+                    payload_retention_days=getattr(ts, "payload_retention_days", 7.0),
+                    ledger_retention_days=getattr(ts, "ledger_retention_days", 90.0),
+                )
+                if any(swept.values()):
+                    log.info("Turn-state TTL sweep: %s", swept)
+            except Exception:
+                pass
+
     def maybe_cleanup(self) -> None:
         """Run cache cleanup if enough time has passed since the last run."""
         try: