fix: Redis outage rate-limit fallback and provider-valid session history

agentkit/api/ratelimit.py

@@ -10,7 +10,7 @@
 
 from fastapi import Depends
 from redis.asyncio import Redis
-from redis.exceptions import ResponseError
+from redis.exceptions import RedisError, ResponseError
 
 from agentkit.api.auth import get_current_settings, verify_api_key
 from agentkit.api.deps import get_redis_client
@@ -83,7 +83,13 @@ async def check(
         limit: int | None = None,
         window_seconds: int | None = None,
     ) -> None:
-        """Validate whether the key has exceeded its rate limit in the current window."""
+        """Validate whether the key has exceeded its rate limit in the current window.
+
+        Redis is authoritative when a client is supplied. Connection failures degrade
+        to the in-memory window rather than failing the request: a process-local limit
+        is stricter than no limit at all, and it keeps the API serving traffic while
+        Redis is unreachable.
+        """
         now = time.time()
         effective_limit = limit or self.requests_per_window or 30
         effective_window = float(window_seconds or self.window_seconds)
@@ -92,42 +98,59 @@ async def check(
             redis_key = f"{self.prefix}{key}"
             member = f"{now}:{uuid.uuid4().hex[:8]}"
 
-            admitted, retry_after = await self._check_window(
-                redis_client=redis_client,
-                redis_key=redis_key,
-                now=now,
-                window=effective_window,
-                limit=effective_limit,
-                member=member,
-            )
-
-            if not admitted:
-                logger.warning(
-                    "Rate limit exceeded for key %s (limit: %d). Retry after: %ds",
-                    key[:8] + "...",
-                    effective_limit,
-                    retry_after,
+            try:
+                admitted, retry_after = await self._check_window(
+                    redis_client=redis_client,
+                    redis_key=redis_key,
+                    now=now,
+                    window=effective_window,
+                    limit=effective_limit,
+                    member=member,
                 )
-                raise RateLimitExceededError(retry_after=retry_after)
-        else:
-            # In-memory sliding window fallback
-            active_timestamps = [
-                t for t in self._memory_store.get(key, []) if t > now - effective_window
-            ]
-
-            if len(active_timestamps) >= effective_limit:
-                oldest_score = active_timestamps[0]
-                retry_after = max(1, math.ceil(oldest_score + effective_window - now))
-                self._memory_store[key] = active_timestamps
+            except RedisError as exc:
                 logger.warning(
-                    "In-memory rate limit exceeded for key %s. Retry after: %ds",
-                    key[:8] + "...",
-                    retry_after,
+                    "Redis unavailable (%s); falling back to in-memory rate limiting.",
+                    exc,
                 )
-                raise RateLimitExceededError(retry_after=retry_after)
+            else:
+                if not admitted:
+                    logger.warning(
+                        "Rate limit exceeded for key %s (limit: %d). Retry after: %ds",
+                        key[:8] + "...",
+                        effective_limit,
+                        retry_after,
+                    )
+                    raise RateLimitExceededError(retry_after=retry_after)
+                return
+
+        self._check_window_in_memory(
+            key=key,
+            now=now,
+            window=effective_window,
+            limit=effective_limit,
+        )
 
-            active_timestamps.append(now)
+    def _check_window_in_memory(self, key: str, now: float, window: float, limit: int) -> None:
+        """Enforce the sliding window in process memory and record the new request.
+
+        Used when Redis is not configured or unreachable. Raises RateLimitExceededError
+        if the window is full; otherwise records ``now`` and returns.
+        """
+        active_timestamps = [t for t in self._memory_store.get(key, []) if t > now - window]
+
+        if len(active_timestamps) >= limit:
+            oldest_score = active_timestamps[0]
+            retry_after = max(1, math.ceil(oldest_score + window - now))
             self._memory_store[key] = active_timestamps
+            logger.warning(
+                "In-memory rate limit exceeded for key %s. Retry after: %ds",
+                key[:8] + "...",
+                retry_after,
+            )
+            raise RateLimitExceededError(retry_after=retry_after)
+
+        active_timestamps.append(now)
+        self._memory_store[key] = active_timestamps
 
     async def _check_window(
         self,

agentkit/core/agent.py

@@ -74,6 +74,46 @@ class RunResult(BaseModel):
     duration_ms: int = Field(default=0, ge=0, description="Execution time in milliseconds.")
 
 
+_ABORTED_TOOL_RESULT_CONTENT = (
+    "ERROR: tool execution was aborted before completion; result unavailable."
+)
+
+
+def _repair_trailing_tool_calls(messages: list[Message]) -> None:
+    """Answer tool calls left unanswered by an aborted run.
+
+    Providers reject histories in which an assistant tool-call turn is not followed by
+    a tool message for every call id (OpenAI returns a 400; Gemini requires a matching
+    function response). A run-level timeout, duplicate-loop abort, or unexpected error
+    can terminate the run after the assistant turn was recorded but before all results
+    were appended. Synthetic error results keep the persisted history valid so the next
+    run in the session can proceed and retry the failed call.
+    """
+    answered: set[str] = set()
+    index = len(messages) - 1
+    while index >= 0 and messages[index].role == Role.TOOL:
+        tool_call_id = messages[index].tool_call_id
+        if tool_call_id is not None:
+            answered.add(tool_call_id)
+        index -= 1
+
+    if index < 0:
+        return
+
+    assistant_message = messages[index]
+    if assistant_message.role != Role.ASSISTANT or not assistant_message.tool_calls:
+        return
+
+    for call in assistant_message.tool_calls:
+        if call.id not in answered:
+            messages.append(
+                Message.tool_result(
+                    tool_call_id=call.id,
+                    content=_ABORTED_TOOL_RESULT_CONTENT,
+                )
+            )
+
+
 class Agent:
     """Core autonomous agent coordinating ReAct reasoning loops."""
 
@@ -113,6 +153,7 @@ async def run(
 
             async def _finalize_and_return(result: RunResult) -> RunResult:
                 if session_id and self.memory is not None and hasattr(self.memory, "save_messages"):
+                    _repair_trailing_tool_calls(messages)
                     with contextlib.suppress(Exception):
                         await self.memory.save_messages(session_id, messages)
 

docs/PROGRESS.md

@@ -59,6 +59,7 @@ This document tracks execution progress across all backlog tasks, including pull
 | AUD-07  | AUD       | Gemini Multi-Turn Role Specification & Active Model Standard | `feat/gemini-tool-calling-fix` | #51 | 2 | Done | Updated function response to role user and default model to gemini-3.1-flash-lite |
 | AUD-08  | AUD       | Expose Configurable Timeout in Docker Compose | `fix/docker-compose-timeout` | #52 | 2 | Done | Exposed RUN_TIMEOUT_S in docker-compose.yml |
 | AUD-09  | AUD       | Live End-to-End Infrastructure Verification & Documentation Proof | `docs/e2e-verification-report` | #53 | 3 | Done | Standardized model, verified live stack, and documented in E2E_REPORT.md |
+| AUD-10  | AUD       | Rate Limiter Outage Fallback & Provider-Valid Session History | `fix/ratelimit-and-session-repair` | #56 | 3 | Done | In-memory fallback on Redis connection failures; synthetic tool results repair aborted sessions |
 
 
 ---
@@ -394,3 +395,9 @@ This document tracks execution progress across all backlog tasks, including pull
 - **Commits**: 3
 - **Notes**: Standardized on `gemini-3.1-flash-lite` across application settings, adapter, factory, compose, and documentation. Authored `docs/E2E_REPORT.md` capturing verbatim live outputs for healthcheck, tool discovery, multi-step analytical SQL join + calculator tax calculation ($3,200 / $3,472), Postgres durable row persistence in `runs` and `steps`, and 7 edge cases (unknown tool, bad arguments, division by zero, max steps exceeded, 429 rate limit with Retry-After, SQL injection blocking, and path traversal blocking). Updated `README.md`.
 
+### AUD-10: Rate Limiter Outage Fallback & Provider-Valid Session History
+- **Status**: Done
+- **PR**: #56
+- **Commits**: 3
+- **Notes**: A full-project review (issue #55) surfaced two medium-severity bugs. `RateLimiter.check` only caught `ResponseError`, so a real Redis outage (`ConnectionError` / `TimeoutError`) escaped the dependency and returned HTTP 500; it now catches `RedisError`, logs a warning, and degrades to the in-memory sliding window, extracted into `_check_window_in_memory`. `Agent._finalize_and_return` could persist histories whose trailing assistant turn requested tool calls with no matching tool messages when a run timed out mid-tool or the duplicate-loop guard aborted mid-batch, which OpenAI and Gemini reject on the next run in that session; `_repair_trailing_tool_calls` now appends synthetic error results for unanswered call ids before saving, on every terminal path. Added regression tests in `tests/unit/test_api_ratelimit.py`, `tests/unit/test_agent_guards.py`, and `tests/unit/test_agent_loop_detector.py`, each confirmed to fail pre-fix.
+

tests/unit/test_agent_guards.py

@@ -6,13 +6,33 @@
 
 from agentkit.core.agent import Agent, AgentConfig
 from agentkit.core.errors import MaxStepsExceeded, RunTimeoutError
-from agentkit.core.types import RunStatus
-from agentkit.llm.base import LLMResponse, TokenUsage
+from agentkit.core.types import Role, RunStatus
+from agentkit.llm.base import LLMResponse, Message, TokenUsage
 from agentkit.llm.fake import FakeLLMClient
+from agentkit.memory.base import InMemoryMemoryStore
 from agentkit.tools.models import ToolCall
 from agentkit.tools.registry import ToolRegistry, tool
 
 
+def _assert_history_is_provider_valid(messages: list[Message]) -> None:
+    """Assert every assistant tool-call id has a matching tool message.
+
+    Mirrors the invariant enforced by OpenAI and Gemini: an assistant turn requesting
+    tool calls must be followed by a tool message for each call id before any other
+    turn appears.
+    """
+    pending: set[str] = set()
+    for msg in messages:
+        if msg.role == Role.TOOL:
+            assert msg.tool_call_id is not None
+            pending.discard(msg.tool_call_id)
+        else:
+            assert not pending, f"Unanswered tool calls before {msg.role} message: {pending}"
+            if msg.role == Role.ASSISTANT:
+                pending = {call.id for call in msg.tool_calls}
+    assert not pending, f"Unanswered tool calls at end of history: {pending}"
+
+
 @pytest.mark.asyncio
 async def test_max_steps_exceeded_returns_status() -> None:
     registry = ToolRegistry()
@@ -166,3 +186,50 @@ async def test_agent_raises_unexpected_exception_when_flag_enabled() -> None:
     agent = Agent(llm=llm, registry=registry)
     with pytest.raises(RuntimeError):
         await agent.run(goal="Handle crash", raise_on_failure=True)
+
+
+@pytest.mark.asyncio
+async def test_run_timeout_persists_provider_valid_session_history() -> None:
+    """A run aborted mid-tool must not leave an unanswered assistant tool-call turn."""
+    registry = ToolRegistry()
+
+    @tool(registry=registry)
+    async def slow_tool_session(delay: float) -> str:
+        """Slow async tool."""
+        await asyncio.sleep(delay)
+        return "done"
+
+    llm = FakeLLMClient(
+        responses=[
+            LLMResponse(
+                text="Calling slow tool",
+                tool_calls=[
+                    ToolCall(id="c_timeout", name="slow_tool_session", arguments={"delay": 0.2})
+                ],
+            )
+        ]
+    )
+    memory = InMemoryMemoryStore()
+    agent = Agent(
+        llm=llm,
+        registry=registry,
+        memory=memory,
+        config=AgentConfig(max_steps=5, run_timeout_s=0.05, tool_timeout_s=1.0),
+    )
+
+    result = await agent.run(goal="Test session timeout", session_id="sess-timeout")
+
+    assert result.status == RunStatus.TIMED_OUT
+
+    messages = await memory.get_messages("sess-timeout")
+    _assert_history_is_provider_valid(messages)
+
+    tool_message = next(m for m in messages if m.tool_call_id == "c_timeout")
+    assert "aborted" in tool_message.content.lower()
+
+    # The session must remain usable for a follow-up run.
+    llm.queue_text("Recovered after timeout.")
+    follow_up = await agent.run(goal="Continue after timeout", session_id="sess-timeout")
+
+    assert follow_up.status == RunStatus.SUCCEEDED
+    assert follow_up.final_answer == "Recovered after timeout."

tests/unit/test_agent_loop_detector.py

@@ -4,13 +4,33 @@
 
 from agentkit.core.agent import Agent, AgentConfig
 from agentkit.core.errors import DuplicateToolCallLoopError
-from agentkit.core.types import RunStatus
-from agentkit.llm.base import LLMResponse
+from agentkit.core.types import Role, RunStatus
+from agentkit.llm.base import LLMResponse, Message
 from agentkit.llm.fake import FakeLLMClient
+from agentkit.memory.base import InMemoryMemoryStore
 from agentkit.tools.models import ToolCall
 from agentkit.tools.registry import ToolRegistry, tool
 
 
+def _assert_history_is_provider_valid(messages: list[Message]) -> None:
+    """Assert every assistant tool-call id has a matching tool message.
+
+    Mirrors the invariant enforced by OpenAI and Gemini: an assistant turn requesting
+    tool calls must be followed by a tool message for each call id before any other
+    turn appears.
+    """
+    pending: set[str] = set()
+    for msg in messages:
+        if msg.role == Role.TOOL:
+            assert msg.tool_call_id is not None
+            pending.discard(msg.tool_call_id)
+        else:
+            assert not pending, f"Unanswered tool calls before {msg.role} message: {pending}"
+            if msg.role == Role.ASSISTANT:
+                pending = {call.id for call in msg.tool_calls}
+    assert not pending, f"Unanswered tool calls at end of history: {pending}"
+
+
 @pytest.mark.asyncio
 async def test_duplicate_tool_call_halts_with_failure_status() -> None:
     registry = ToolRegistry()
@@ -150,3 +170,87 @@ def tool_b(x: int) -> int:
 
     assert result.status == RunStatus.SUCCEEDED
     assert result.final_answer == "Done"
+
+
+@pytest.mark.asyncio
+async def test_duplicate_abort_persists_provider_valid_session_history() -> None:
+    """The unexecuted duplicate call must still get a tool message in the session."""
+    registry = ToolRegistry()
+
+    @tool(registry=registry)
+    def repeat_session(val: int) -> int:
+        """Repeat."""
+        return val
+
+    first_call = ToolCall(id="c1", name="repeat_session", arguments={"val": 42})
+    duplicate_call = ToolCall(id="c2", name="repeat_session", arguments={"val": 42})
+    llm = FakeLLMClient(
+        responses=[
+            LLMResponse(text="Repeat 1", tool_calls=[first_call]),
+            LLMResponse(text="Repeat 2", tool_calls=[duplicate_call]),
+        ]
+    )
+    memory = InMemoryMemoryStore()
+    agent = Agent(
+        llm=llm,
+        registry=registry,
+        memory=memory,
+        config=AgentConfig(max_steps=5, max_consecutive_duplicate_tool_calls=2),
+    )
+
+    result = await agent.run(goal="Loop with session", session_id="sess-loop")
+
+    assert result.status == RunStatus.FAILED
+
+    messages = await memory.get_messages("sess-loop")
+    _assert_history_is_provider_valid(messages)
+
+    tool_message = next(m for m in messages if m.tool_call_id == "c2")
+    assert "aborted" in tool_message.content.lower()
+
+
+@pytest.mark.asyncio
+async def test_duplicate_abort_mid_batch_answers_remaining_calls() -> None:
+    """Aborting partway through a multi-call turn must not drop the later call ids."""
+    registry = ToolRegistry()
+
+    @tool(registry=registry)
+    def echo_tool(v: int) -> int:
+        """Echo value."""
+        return v
+
+    llm = FakeLLMClient(
+        responses=[
+            LLMResponse(
+                text="Setup call",
+                tool_calls=[ToolCall(id="c_setup", name="echo_tool", arguments={"v": 1})],
+            ),
+            LLMResponse(
+                text="Batch with a repeat",
+                tool_calls=[
+                    ToolCall(id="c_batch_1", name="echo_tool", arguments={"v": 2}),
+                    ToolCall(id="c_batch_2", name="echo_tool", arguments={"v": 2}),
+                ],
+            ),
+        ]
+    )
+    memory = InMemoryMemoryStore()
+    agent = Agent(
+        llm=llm,
+        registry=registry,
+        memory=memory,
+        config=AgentConfig(max_steps=5, max_consecutive_duplicate_tool_calls=2),
+    )
+
+    result = await agent.run(goal="Loop mid batch", session_id="sess-mid-batch")
+
+    assert result.status == RunStatus.FAILED
+
+    messages = await memory.get_messages("sess-mid-batch")
+    _assert_history_is_provider_valid(messages)
+
+    executed = next(m for m in messages if m.tool_call_id == "c_batch_1")
+    assert executed.content == "2"
+
+    interrupted = next(m for m in messages if m.tool_call_id == "c_batch_2")
+    assert "aborted" in interrupted.content.lower()