diff --git a/roboco/api/middleware.py b/roboco/api/middleware.py index a97cfc58..1044b90e 100644 --- a/roboco/api/middleware.py +++ b/roboco/api/middleware.py @@ -4,6 +4,8 @@ API Middleware Request/response middleware for logging, error handling, and correlation IDs. """ +import asyncio +import json import time import uuid from collections.abc import Callable, Sequence @@ -16,8 +18,10 @@ from fastapi.encoders import jsonable_encoder from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware +from starlette.types import ASGIApp, Receive, Scope, Send from roboco.api.schemas.common import ErrorCode +from roboco.config import settings from roboco.exceptions import ( AuthenticationError, InvalidStateError, @@ -472,6 +476,91 @@ def setup_middleware(app: FastAPI) -> None: app.add_exception_handler(RateLimitError, rate_limit_exception_handler) app.add_exception_handler(Exception, generic_exception_handler) - # Middleware (added in reverse order due to LIFO) + # Middleware (added in reverse order due to LIFO): the LAST add_middleware + # call is the OUTERMOST. FlowVerbTimeoutMiddleware is added FIRST so it is + # the INNERMOST — closest to the routes — meaning correlation + logging + # still wrap the 504 it returns, AND its asyncio.timeout cancels the route + # coroutine + its get_db dependency directly (same task, reliable cancel). + app.add_middleware(FlowVerbTimeoutMiddleware) app.add_middleware(RequestLoggingMiddleware) app.add_middleware(CorrelationIdMiddleware) + + +class FlowVerbTimeoutMiddleware: + """Pure-ASGI server-side timeout on gateway intent-verb requests. + + A hung flow verb — e.g. ``claim()`` blocked on a ``SELECT ... FOR + UPDATE`` row lock held by a prior stuck transaction — would otherwise hold + its request transaction open indefinitely. uvicorn does not cancel the + endpoint coroutine on client disconnect, and ``get_db`` only rolled back + on ``Exception``, so the row lock was never released and every later + task-row write on that task wedged (the 2026-07-07 + ``kimi-k2.7-code:cloud`` agent on task 79d686f0). This wraps each + ``/api/v1/flow/*`` request in ``asyncio.timeout``; on expiry the inner + app is cancelled (CancelledError propagates through ``get_db``, which now + rolls back, releasing the lock) and a clean retryable 504 envelope is + returned. Pure ASGI (not BaseHTTPMiddleware) so cancellation propagates + into the route coroutine without the spawned-task gap. + + Reads (``evidence``) and journal writes (``note``) don't touch the task + row, so they are unaffected; only task-row writes route through ``claim``. + """ + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http" or not scope["path"].startswith("/api/v1/flow/"): + await self.app(scope, receive, send) + return + timeout = settings.flow_verb_timeout_seconds + started = False + + async def send_wrapper(message: Any) -> None: + nonlocal started + if message["type"] == "http.response.start": + started = True + await send(message) + + try: + async with asyncio.timeout(timeout): + await self.app(scope, receive, send_wrapper) + except TimeoutError: + # The inner app was cancelled mid-verb; get_db has already rolled + # back (releasing the FOR UPDATE lock) by the time we get here. + if started: + # The route had already begun a response before the timeout + # fired — the client owns whatever was sent; we cannot start + # a new one. Rare for a hung verb (it hangs before responding). + return + body = json.dumps( + { + "status": None, + "task_id": None, + "next": None, + "evidence": {}, + "context_briefing": {}, + "error": "gateway_timeout", + "message": ( + f"verb exceeded the {timeout:.0f}s server-side timeout; " + "the request transaction was rolled back" + ), + "remediate": ( + "retry the verb; if it persists the underlying task " + "may be wedged — escalate" + ), + } + ).encode() + headers = [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode()), + ] + try: + await send( + {"type": "http.response.start", "status": 504, "headers": headers} + ) + await send({"type": "http.response.body", "body": body}) + except Exception: + # Client may have already disconnected (the original trigger); + # the lock is released regardless. Nothing to do. + pass diff --git a/roboco/config.py b/roboco/config.py index 25ea60d1..aeeb6c15 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -1288,6 +1288,22 @@ class Settings(BaseSettings): "design — most git operations are sub-second." ), ) + flow_verb_timeout_seconds: float = Field( + default=120.0, + ge=1.0, + description=( + "Server-side wall-clock timeout for a single gateway intent-verb " + "request (/api/v1/flow/*). A verb whose transaction hangs — e.g. " + "claim() blocked on a FOR UPDATE row lock held by a prior stuck " + "transaction — would otherwise hold its request transaction open " + "indefinitely: uvicorn does not cancel the endpoint coroutine on " + "client disconnect, so the row lock is never released and every " + "later task-row write on that task wedges. On expiry the inner " + "app is cancelled, get_db rolls back (releasing the lock), and a " + "retryable 504 envelope is returned. Generous by default so " + "legitimate verbs are unaffected." + ), + ) git_commit_timeout_seconds: int = Field( default=180, ge=30, diff --git a/roboco/db/base.py b/roboco/db/base.py index 620da9af..eb09cd31 100644 --- a/roboco/db/base.py +++ b/roboco/db/base.py @@ -88,7 +88,12 @@ async def get_db() -> AsyncGenerator[AsyncSession]: try: yield session await session.commit() - except Exception: + except (Exception, asyncio.CancelledError): + # CancelledError is BaseException, so the bare `except Exception` + # did not catch it — a server-side asyncio.timeout cancelling a + # hung verb (FlowVerbTimeoutMiddleware) would otherwise leave the + # request transaction unrolled-back, holding its FOR UPDATE row + # lock. Roll back on cancellation too so the lock releases. await session.rollback() raise @@ -107,7 +112,12 @@ async def get_db_context() -> AsyncGenerator[AsyncSession]: try: yield session await session.commit() - except Exception: + except (Exception, asyncio.CancelledError): + # CancelledError is BaseException, so the bare `except Exception` + # did not catch it — a server-side asyncio.timeout cancelling a + # hung verb (FlowVerbTimeoutMiddleware) would otherwise leave the + # request transaction unrolled-back, holding its FOR UPDATE row + # lock. Roll back on cancellation too so the lock releases. await session.rollback() raise diff --git a/tests/e2e_smoke/test_flow_verb_timeout.py b/tests/e2e_smoke/test_flow_verb_timeout.py new file mode 100644 index 00000000..de6b6288 --- /dev/null +++ b/tests/e2e_smoke/test_flow_verb_timeout.py @@ -0,0 +1,240 @@ +"""Scenario: a hung gateway flow-verb must release its task-row FOR UPDATE +lock within a bounded server-side timeout — not hold it forever. + +Reproduces the 2026-07-07 wedge (a ``kimi-k2.7-code:cloud`` agent on task +79d686f0): a verb whose request transaction held the ``SELECT ... FOR +UPDATE`` lock on the task row never committed. uvicorn does not cancel the +endpoint coroutine on client disconnect, and ``get_db`` only rolled back on +``Exception`` (not a hang), so the row lock was held indefinitely — every +later task-row write on that task blocked on the lock and timed out, while +plain reads (``evidence``) and journal writes (``note``, a different row) +stayed fast. The fix is a server-side ``asyncio.timeout`` on flow verbs +(``FlowVerbTimeoutMiddleware``): on expiry the inner app is cancelled, +``get_db`` rolls back (releasing the lock), and a clean retryable 504 +``gateway_timeout`` envelope is returned. + +The hang is injected INSIDE the verb's own transaction: ``claim`` acquires +the FOR UPDATE lock, then ``set_plan`` sleeps past the server timeout. The +sleep fires on the FIRST ``set_plan`` call only — a retry ``i_will_plan`` +short-circuits as idempotent re-entry (the umbrella scenario's claim-time +gate dance relies on the same behavior: the composed sequence runs once, +retries skip it), so the hang has to be on the one call that actually runs. + +Two tests give the empirical proof: + +- ``test_flow_verb_timeout_releases_row_lock`` (fix ARMED, server timeout + 1s): verb-1's hang is cancelled at 1s, ``get_db`` rolls back (lock + released), a 504 ``gateway_timeout`` comes back, and verb-2 re-acquires + the row and reaches the post-claim gate (``tracing_gap``). Reaching the + gate is the success marker — the verb completed its composed transaction. +- ``test_flow_verb_holds_lock_when_timeout_disarmed`` (fix DISARMED, server + timeout 1000s + MCP client timeout 3s): verb-1's hang is NOT cancelled, + so it holds the FOR UPDATE lock past the client's HTTP timeout — the + empirical reproduction of the wedge on the same branch, by turning the + fix off. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any + +import httpx +from tests.e2e_smoke.arcs import ( + origin_branch, + seed_company, + seed_project, + seed_task, + set_branch_name, +) +from tests.e2e_smoke.harness import ScriptedAgent, expect_error + +if TYPE_CHECKING: + from uuid import UUID + + import pytest + from tests.e2e_smoke.arcs import Company + from tests.e2e_smoke.harness import E2EStack + +# Pydantic's IWillPlanRequest.approach enforces >= 150 chars at the HTTP +# boundary, so every i_will_plan call needs a compliant approach. +_APPROACH = ( + "Plan and delegate the page-scoped refresh button work to the frontend " + "cell: land the provider/hook, add the navbar button, remove the inline " + "buttons, and route one planning subtask to fe-pm for delivery." +) +_SUB_TASKS = [ + { + "title": "Frontend cell: refresh button", + "description": ( + "Delegate the navbar refresh button to fe-pm: land the " + "provider/hook and wire the click handler into the page." + ), + } +] +_PLAN = "Land the refresh button via the frontend cell." +# set_plan sleeps this long inside the verb's own transaction. On the fix +# (1s server timeout) the sleep is cancelled well before this; disarmed +# (1000s server timeout, 3s client timeout) the client trips first. +_HANG_SECONDS = 8.0 +_SERVER_TIMEOUT_SECONDS = 1.0 +_DISARMED_SERVER_TIMEOUT_SECONDS = 1000.0 +# The MCP client's HTTP timeout for the disarmed reproduction — must be less +# than _HANG_SECONDS so the client trips before the sleep ends. Applied +# AFTER priming the per-agent flow_server reload (a pre-call patch is +# clobbered by the reload on the first flow() call). +_MCP_CLIENT_TIMEOUT_SECONDS = 3.0 + + +def _seed_planning_root( + stack: E2EStack, company: Company +) -> tuple[ScriptedAgent, UUID]: + """Seed a PENDING MAIN_PM planning root + its origin branch for the test.""" + from roboco.models import Team + from roboco.models.base import TaskStatus, TaskType + + project_id, _project_slug = seed_project(stack, company) + main_pm = ScriptedAgent(stack, company.main_pm_id, "main-pm", "main_pm") + task_id = seed_task( + stack, + title="Root: page-scoped refresh button", + description="Frontend-only root: provider/hook + navbar button.", + acceptance_criteria=["the refresh button lands on master"], + task_type=TaskType.PLANNING, + team=Team.MAIN_PM, + project_id=project_id, + created_by=company.main_pm_id, + assigned_to=company.main_pm_id, + status=TaskStatus.PENDING, + ) + branch = f"feature/main_pm/{str(task_id)[:8]}" + origin_branch(stack, branch, start="master") + set_branch_name(stack, task_id, branch) + return main_pm, task_id + + +def _patch_hang_in_set_plan( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Make the first ``TaskService.set_plan`` call sleep past the timeout. + + The hang is inside the verb's own transaction: claim has already + acquired the FOR UPDATE row lock when set_plan runs, so the sleep holds + the lock until the transaction commits or rolls back. Only the FIRST + call sleeps — a retry i_will_plan short-circuits as re-entry and never + re-runs set_plan, so a per-call counter would never reach a second hang. + """ + from roboco.services.task import TaskService + + real_set_plan = TaskService.set_plan + hung = {"done": False} + + async def hung_set_plan( + self: TaskService, task_id: UUID, plan: str | dict[str, Any] + ) -> Any: + if not hung["done"]: + hung["done"] = True + await asyncio.sleep(_HANG_SECONDS) + return await real_set_plan(self, task_id, plan) + + monkeypatch.setattr(TaskService, "set_plan", hung_set_plan) + + +def test_flow_verb_timeout_releases_row_lock( + e2e_stack: E2EStack, monkeypatch: pytest.MonkeyPatch +) -> None: + """Fix ARMED: a hung verb is cancelled, its lock released, retry proceeds.""" + from roboco.config import settings + + stack = e2e_stack + company = seed_company(stack) + main_pm, task_id = _seed_planning_root(stack, company) + + # Server-side verb timeout: 1s. A hung verb is cancelled and its + # transaction rolled back at this boundary instead of holding the FOR + # UPDATE lock forever. + monkeypatch.setattr(settings, "flow_verb_timeout_seconds", _SERVER_TIMEOUT_SECONDS) + _patch_hang_in_set_plan(monkeypatch) + + # 1. verb-1: claim -> FOR UPDATE lock -> set_plan HANGS. The server + # cancels at 1s, get_db rolls back (releasing the lock), and a 504 + # gateway_timeout envelope comes back within the default client + # window (the fix makes the hang bounded; no client-side race). + env1 = main_pm.flow( + "i_will_plan", + task_id=str(task_id), + plan=_PLAN, + approach=_APPROACH, + sub_tasks=_SUB_TASKS, + ) + expect_error(env1, "gateway_timeout", "verb-1 must return a bounded 504, not hang") + + # 2. verb-2: the task was rolled back to PENDING by verb-1's + # cancellation, so claim re-enters and acquires the row (only + # possible because verb-1 released the lock). set_plan runs for + # real, start commits, and the post-claim tracing gate fires (no + # decision note) -> tracing_gap. Reaching the gate is the success + # marker: on master the row would still be locked by verb-1's stuck + # coroutine and this claim would hang on the FOR UPDATE. + env2 = main_pm.flow( + "i_will_plan", + task_id=str(task_id), + plan=_PLAN, + approach=_APPROACH, + sub_tasks=_SUB_TASKS, + ) + expect_error( + env2, + "tracing_gap", + "verb-2 reaches the post-claim gate — the row lock was released", + ) + + +def test_flow_verb_holds_lock_when_timeout_disarmed( + e2e_stack: E2EStack, monkeypatch: pytest.MonkeyPatch +) -> None: + """Fix DISARMED: with no bounded server timeout, the hung verb holds the + lock past the client's HTTP timeout — the empirical reproduction of the + 2026-07-07 wedge on the same branch, by turning the fix off. + """ + from roboco.config import settings + from roboco.mcp import flow_server + + stack = e2e_stack + company = seed_company(stack) + main_pm, task_id = _seed_planning_root(stack, company) + + # Disarm the server-side timeout: 1000s. The hung verb is NOT cancelled + # by the server, so it holds the FOR UPDATE lock for the full sleep. + monkeypatch.setattr( + settings, "flow_verb_timeout_seconds", _DISARMED_SERVER_TIMEOUT_SECONDS + ) + _patch_hang_in_set_plan(monkeypatch) + + # Prime the per-agent flow_server reload with a no-op verb (i_am_idle + # touches no task). The reload resets module globals (_TIMEOUT=30), so a + # pre-call patch would be clobbered; after this call the module is + # pinned to this agent and the patch below survives. + main_pm.flow("i_am_idle") + monkeypatch.setattr(flow_server, "_TIMEOUT", _MCP_CLIENT_TIMEOUT_SECONDS) + + # verb-1: claim -> FOR UPDATE lock -> set_plan HANGS. The server does + # not cancel (1000s timeout), so the verb holds the lock past the + # client's 3s HTTP timeout — the wedge. httpx raises, never returning + # an envelope. + verb1_hung = False + try: + main_pm.flow( + "i_will_plan", + task_id=str(task_id), + plan=_PLAN, + approach=_APPROACH, + sub_tasks=_SUB_TASKS, + ) + except httpx.TimeoutException: + verb1_hung = True + assert verb1_hung, ( + "verb-1 did NOT hang with the server timeout disarmed — the wedge is " + "not reproduced; either the hang injection broke or the server is " + "cancelling despite the disarmed timeout" + )