From 92ab13bce0177c7ee80f626d5006f106b8ace773 Mon Sep 17 00:00:00 2001 From: Renzo F <45401804+rennf93@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:02:29 +0200 Subject: [PATCH] Fix/backend/flow verb timeout row lock (#326) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(sandbox): pluggable per-engine registry (postgres/redis/mongo) Replaces the hardcoded postgres+redis branches in the provisioner and the env emitter with a registry of SandboxEngine specs (image, run args, readiness probe, connection, ROBOCO_TEST_* env) in a pure low module (roboco/models/sandbox.py). VALID_SANDBOX_SERVICES is derived from the registry — single source of truth — and the provisioner + orchestrator iterate it, so adding an engine is one class + one registry line, not another branch. Adds a mongo:8-alpine engine (ROBOCO_TEST_MONGO_*) as the third service alongside postgres/redis. Also fixes the cold-pull loop that stranded v0.19.0 board agents with empty error strings: docker run pulled inline under a 20s deadline, so a NAS cold pull was killed, cancelled, and re-pulled from scratch forever. _ensure_image now inspects + pulls (300s) before run; provisioning errors log type+message so a bare TimeoutError no longer shows as "". Panel edit-project dialog: postgres/redis toggles -> a Set multi-select driven by a SANDBOX_SERVICES catalog, so new engines appear in the UI by adding to the catalog. Tests: engine parity (allowlist==registry, unique slugs/images, no None leak in env, SandboxInfo aggregates every engine), mongo provision + env injection, plus the existing postgres/redis provision/env/spawn/janitor suite updated to the registry shape. 821 unit / 5 skip green; ruff + mypy (360 files) clean. * docs(sandbox): reflect pluggable engine registry + mongo across docs CHANGELOG (0.19.0): Added entry for the pluggable sandbox engine registry (postgres/redis/mongo) + Fixed entry for the cold-pull loop/empty-error strand that boarded v0.19.0 board agents. docs/map (9 files): sandbox subsystem blurbs, SandboxProvisioner rows, _maybe_provision_sandbox/_append_sandbox_env rows, feature-flag rows, the migration-057 row + v0.17.0 delta, and the models.md VALID_SANDBOX_SERVICES note — all retitled to DB/Redis/Mongo via the engine registry (roboco/models/sandbox.py), with the one-class-one-line extension story and the _ensure_image cold-pull fix. Production-network (roboco_data) lines left as postgres+redis — mongo is sandbox-only, not a prod service. docs/rag (3 files): sandbox-db.md rewritten around the registry (engine list, generic _provision_engine, image pre-pull, ROBOCO_TEST_DB_*/REDIS_*/MONGO_* incl. MONGO_AUTH_DB=admin, single emit_env); config-reference sandbox flag row + subsection retitled; db-network-isolation framing broadened to postgres/redis/mongo. preconditions-and-rejections left untouched (its hit was an unrelated gateway see-also link). * test(e2e): harden umbrella close terminal reads with bounded wait-for-state The MegaTask umbrella close test flaked once on CI (ceo-approve returned 200 but the re-fetch saw awaiting_pm_review) then passed on re-run. The production path is deterministic: complete -> main_pm_complete -> submit_pm_review -> escalate_to_ceo -> ceo_approve -> commit, all on one session, all awaited; the fire-and-forget completion hooks are isolated (own session, best-effort, never touch task.status or the request session). 20 local runs could not reproduce it. The one real surface is the read pattern: the e2e stack commits on the uvicorn thread's loop and reads via a separate loop (run_db -> asyncio.run with a fresh engine), so a terminal single point-read can race a still-draining completion hook on a contended runner. Replace the two terminal point-reads with a bounded wait_for_status poll. Strictly better than a one-shot read: absorbs the transient, and a genuine state bug still surfaces via the timeout branch asserting against the last-read state. * fix(gateway): bound hung flow-verbs with a server-side timeout A gateway intent-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/cancellation), so the row lock was held indefinitely and every later task-row write on that task wedged (2026-07-07 kimi-k2.7-code:cloud agent on task 79d686f0). Reads (evidence) and journal writes (note) stayed fast — the symptom that pointed at a task-row lock. Fix: pure-ASGI FlowVerbTimeoutMiddleware wraps each /api/v1/flow/* request in asyncio.timeout(flow_verb_timeout_seconds, default 120s). On expiry the inner app is cancelled; CancelledError now propagates through get_db (which catches it alongside Exception and rolls back), releasing the FOR UPDATE lock, and a retryable 504 gateway_timeout envelope is returned. Pure ASGI (not BaseHTTPMiddleware) so cancellation reaches the route coroutine + get_db dependency directly, with no spawned-task gap. Registered innermost so correlation + logging still wrap the 504. E2E: two fault-injection scenarios in tests/e2e_smoke/test_flow_verb_timeout.py. A hang is injected inside the verb's own transaction (claim acquires the FOR UPDATE lock, then set_plan sleeps past the timeout; only the first set_plan call runs — a retry short-circuits as idempotent re-entry). - ARMED (server timeout 1s): verb-1 returns a bounded 504 gateway_timeout, verb-2 re-acquires the row and reaches the post-claim gate (tracing_gap) — proving the lock was released by verb-1's cancellation. - DISARMED (server timeout 1000s, MCP client timeout 3s): verb-1 holds the lock past the client's HTTP timeout — the empirical reproduction of the wedge on the same branch, by turning the fix off. Full e2e suite green (32 passed). * feat(sandbox): pluggable per-engine registry (postgres/redis/mongo) (#324) (#325) * feat(sandbox): pluggable per-engine registry (postgres/redis/mongo) Replaces the hardcoded postgres+redis branches in the provisioner and the env emitter with a registry of SandboxEngine specs (image, run args, readiness probe, connection, ROBOCO_TEST_* env) in a pure low module (roboco/models/sandbox.py). VALID_SANDBOX_SERVICES is derived from the registry — single source of truth — and the provisioner + orchestrator iterate it, so adding an engine is one class + one registry line, not another branch. Adds a mongo:8-alpine engine (ROBOCO_TEST_MONGO_*) as the third service alongside postgres/redis. Also fixes the cold-pull loop that stranded v0.19.0 board agents with empty error strings: docker run pulled inline under a 20s deadline, so a NAS cold pull was killed, cancelled, and re-pulled from scratch forever. _ensure_image now inspects + pulls (300s) before run; provisioning errors log type+message so a bare TimeoutError no longer shows as "". Panel edit-project dialog: postgres/redis toggles -> a Set multi-select driven by a SANDBOX_SERVICES catalog, so new engines appear in the UI by adding to the catalog. Tests: engine parity (allowlist==registry, unique slugs/images, no None leak in env, SandboxInfo aggregates every engine), mongo provision + env injection, plus the existing postgres/redis provision/env/spawn/janitor suite updated to the registry shape. 821 unit / 5 skip green; ruff + mypy (360 files) clean. * docs(sandbox): reflect pluggable engine registry + mongo across docs CHANGELOG (0.19.0): Added entry for the pluggable sandbox engine registry (postgres/redis/mongo) + Fixed entry for the cold-pull loop/empty-error strand that boarded v0.19.0 board agents. docs/map (9 files): sandbox subsystem blurbs, SandboxProvisioner rows, _maybe_provision_sandbox/_append_sandbox_env rows, feature-flag rows, the migration-057 row + v0.17.0 delta, and the models.md VALID_SANDBOX_SERVICES note — all retitled to DB/Redis/Mongo via the engine registry (roboco/models/sandbox.py), with the one-class-one-line extension story and the _ensure_image cold-pull fix. Production-network (roboco_data) lines left as postgres+redis — mongo is sandbox-only, not a prod service. docs/rag (3 files): sandbox-db.md rewritten around the registry (engine list, generic _provision_engine, image pre-pull, ROBOCO_TEST_DB_*/REDIS_*/MONGO_* incl. MONGO_AUTH_DB=admin, single emit_env); config-reference sandbox flag row + subsection retitled; db-network-isolation framing broadened to postgres/redis/mongo. preconditions-and-rejections left untouched (its hit was an unrelated gateway see-also link). * test(e2e): harden umbrella close terminal reads with bounded wait-for-state The MegaTask umbrella close test flaked once on CI (ceo-approve returned 200 but the re-fetch saw awaiting_pm_review) then passed on re-run. The production path is deterministic: complete -> main_pm_complete -> submit_pm_review -> escalate_to_ceo -> ceo_approve -> commit, all on one session, all awaited; the fire-and-forget completion hooks are isolated (own session, best-effort, never touch task.status or the request session). 20 local runs could not reproduce it. The one real surface is the read pattern: the e2e stack commits on the uvicorn thread's loop and reads via a separate loop (run_db -> asyncio.run with a fresh engine), so a terminal single point-read can race a still-draining completion hook on a contended runner. Replace the two terminal point-reads with a bounded wait_for_status poll. Strictly better than a one-shot read: absorbs the transient, and a genuine state bug still surfaces via the timeout branch asserting against the last-read state. --------- Co-authored-by: Renn F --------- Co-authored-by: Renn F --- roboco/api/middleware.py | 91 +++++++- roboco/config.py | 16 ++ roboco/db/base.py | 14 +- tests/e2e_smoke/test_flow_verb_timeout.py | 240 ++++++++++++++++++++++ 4 files changed, 358 insertions(+), 3 deletions(-) create mode 100644 tests/e2e_smoke/test_flow_verb_timeout.py 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" + )