From f56b1bcd60f01a1bc72d09f0c2a58b117f72516c Mon Sep 17 00:00:00 2001 From: Amar K P Date: Sun, 18 Oct 2026 22:11:25 +0530 Subject: [PATCH 1/3] fix(api): fall back to in-memory rate limiting when Redis is unreachable - Catch RedisError (not just ResponseError) around the sliding-window script so connection failures degrade to the in-memory window instead of surfacing as 500 - Extract the in-memory window into _check_window_in_memory shared by both paths - Add unit and API-level regression tests for an unreachable Redis client --- agentkit/api/ratelimit.py | 89 ++++++++++++++++++++------------ tests/unit/test_api_ratelimit.py | 55 +++++++++++++++++++- 2 files changed, 110 insertions(+), 34 deletions(-) diff --git a/agentkit/api/ratelimit.py b/agentkit/api/ratelimit.py index 14d0220..610662b 100644 --- a/agentkit/api/ratelimit.py +++ b/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, diff --git a/tests/unit/test_api_ratelimit.py b/tests/unit/test_api_ratelimit.py index c347ad8..35803f9 100644 --- a/tests/unit/test_api_ratelimit.py +++ b/tests/unit/test_api_ratelimit.py @@ -1,13 +1,14 @@ from __future__ import annotations from collections.abc import AsyncIterator -from typing import Any +from typing import Any, cast from unittest.mock import patch import fakeredis.aioredis import pytest from httpx import ASGITransport, AsyncClient from redis.asyncio import Redis +from redis.exceptions import ConnectionError as RedisConnectionError from agentkit.api.main import create_app from agentkit.api.ratelimit import RateLimiter @@ -27,6 +28,13 @@ async def fake_redis() -> AsyncIterator[Redis[Any]]: await client.close() +class UnreachableRedis: + """Minimal Redis stand-in whose connection is always refused.""" + + async def eval(self, *_args: Any, **_kwargs: Any) -> Any: + raise RedisConnectionError("Connection refused") + + @pytest.mark.asyncio async def test_rate_limiter_allows_under_limit_redis(fake_redis: Redis[Any]) -> None: limiter = RateLimiter(requests_per_window=3, window_seconds=60) @@ -113,3 +121,48 @@ async def test_rate_limit_api_endpoint_429_with_headers(fake_redis: Redis[Any]) assert "Retry-After" in res3.headers data = res3.json() assert data["error"]["code"] == "RATE_LIMIT_EXCEEDED" + + +@pytest.mark.asyncio +async def test_rate_limiter_falls_back_to_memory_when_redis_unreachable() -> None: + """A refused Redis connection degrades to the in-memory window instead of raising.""" + limiter = RateLimiter(requests_per_window=2, window_seconds=10) + key = "ak_unreachable_key" + unreachable = cast("Redis[Any]", UnreachableRedis()) + + base_time = 3000.0 + with patch("time.time", return_value=base_time): + await limiter.check(key=key, redis_client=unreachable) + await limiter.check(key=key, redis_client=unreachable) + with pytest.raises(RateLimitExceededError) as exc_info: + await limiter.check(key=key, redis_client=unreachable) + assert exc_info.value.retry_after > 0 + + with patch("time.time", return_value=base_time + 11.0): + await limiter.check(key=key, redis_client=unreachable) + + +@pytest.mark.asyncio +async def test_rate_limit_endpoint_degrades_to_memory_during_redis_outage() -> None: + """Endpoints keep serving and enforcing limits while Redis is unreachable.""" + test_settings = Settings( + API_KEYS="test_api_key_unreachable", + RATE_LIMIT_PER_MIN=2, + ) + app = create_app(settings=test_settings) + app.state.redis_client = cast("Redis[Any]", UnreachableRedis()) + + transport = ASGITransport(app=app) + headers = {"X-API-Key": "test_api_key_unreachable"} + + async with AsyncClient(transport=transport, base_url="http://test") as client: + res1 = await client.get("/tools", headers=headers) + assert res1.status_code == 200 + + res2 = await client.get("/tools", headers=headers) + assert res2.status_code == 200 + + res3 = await client.get("/tools", headers=headers) + assert res3.status_code == 429 + assert "Retry-After" in res3.headers + assert res3.json()["error"]["code"] == "RATE_LIMIT_EXCEEDED" From acac87bb826fc554ba21fb456b1f7f3bef466eb1 Mon Sep 17 00:00:00 2001 From: Amar K P Date: Sun, 18 Oct 2026 22:11:29 +0530 Subject: [PATCH 2/3] fix(core): persist provider-valid session history after aborted runs - Repair the trailing assistant tool-call turn before saving session messages so timeouts, duplicate-loop aborts, and unexpected errors cannot leave dangling tool_calls that OpenAI and Gemini reject on the next run - Append synthetic error tool results for unanswered call ids, preserving any partial results already recorded - Add regression tests covering timeout, duplicate abort, and mid-batch abort --- agentkit/core/agent.py | 41 ++++++++++ tests/unit/test_agent_guards.py | 71 +++++++++++++++- tests/unit/test_agent_loop_detector.py | 108 ++++++++++++++++++++++++- 3 files changed, 216 insertions(+), 4 deletions(-) diff --git a/agentkit/core/agent.py b/agentkit/core/agent.py index 5f2c1c3..94b67f1 100644 --- a/agentkit/core/agent.py +++ b/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) diff --git a/tests/unit/test_agent_guards.py b/tests/unit/test_agent_guards.py index 405a6c2..b441ba4 100644 --- a/tests/unit/test_agent_guards.py +++ b/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." diff --git a/tests/unit/test_agent_loop_detector.py b/tests/unit/test_agent_loop_detector.py index bfc8943..fbed2ca 100644 --- a/tests/unit/test_agent_loop_detector.py +++ b/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() From 1b6ccc1fe3ad50d0f43de1fa8f69fd8fe011f8d3 Mon Sep 17 00:00:00 2001 From: Amar K P Date: Sun, 18 Oct 2026 22:12:07 +0530 Subject: [PATCH 3/3] docs(progress): record AUD-10 resolution and PR link --- docs/PROGRESS.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index 157574e..2af443f 100644 --- a/docs/PROGRESS.md +++ b/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. +