From 5e6bfa53406c83635fc489c19d7241da3ccc5494 Mon Sep 17 00:00:00 2001 From: OfficialAbhinavSingh Date: Fri, 6 Nov 2026 11:58:13 +0530 Subject: [PATCH 1/2] control-plane, frontend: retries no longer die silently; a stuck task can be retried A retry after an attempt that ended without a receipt never ran. The gateway holds attempt n while attempt n-1 still holds its payment (503), and that payment only comes back at the attempt's deadline, 30 minutes after dispatch. The retry's dispatch gave up after event_retry_window_s (10 minutes) and nothing read it again, so the task said "Retrying" forever, and every later event of the plan, Close plan included, waited behind it. - A retry's dispatch keeps being delivered for the deadline plus 10 minutes (dispatch.RETRY_WINDOW_S), and while it waits it holds up no other event of its plan. - Closing a plan cancels its undelivered dispatches, so nothing locks money after the requester gave up. - stuck_indexes counts the newest dispatch: one still on its way is in flight, one given up waits on the requester. - POST /v0/plans/{uuid}/tasks/{index}/retry {agent_id}: a stuck task goes to an agent the requester chooses, paid from that task's own escrow, up to max_requester_attempts (6) attempts in all. The run page offers it next to Close plan. - The plan detail returns pending_dispatches and stuck_indexes. The run page says what a retry waits on and reloads while one is pending, since the stream carries jobs, not dispatches. - The worker refuses to start without its sandbox image, so jobs stay queued instead of each failing and costing the requester an agent pick. --- control-plane/mergit_cp/config.py | 2 + control-plane/mergit_cp/dispatch.py | 11 +- control-plane/mergit_cp/events.py | 6 +- control-plane/mergit_cp/outcomes.py | 90 +++++++++--- control-plane/mergit_cp/routes_plans.py | 42 ++++++ control-plane/mergit_cp/routes_runs.py | 8 +- control-plane/mergit_cp/runs.py | 51 +++++++ control-plane/mergit_worker/__main__.py | 9 +- control-plane/mergit_worker/sandbox.py | 18 ++- control-plane/tests/test_api.py | 45 ++++++ control-plane/tests/test_events.py | 53 +++++++ control-plane/tests/test_outcomes.py | 133 ++++++++++++++++++ control-plane/tests/test_sandbox_image.py | 36 +++++ frontend/e2e/live/plans.ts | 30 ++++ frontend/e2e/live/run.spec.ts | 29 +++- frontend/src/api/client.ts | 8 ++ frontend/src/api/types.ts | 20 ++- .../src/screens/live/run/LiveRunWorkspace.tsx | 23 ++- frontend/src/screens/live/run/RetryTask.tsx | 64 +++++++++ frontend/src/screens/live/run/RunView.tsx | 33 +++-- frontend/src/screens/live/run/TaskDetail.tsx | 22 ++- frontend/src/screens/live/run/TaskFlow.tsx | 4 +- .../src/screens/live/run/progress.test.ts | 39 ++++- frontend/src/screens/live/run/progress.ts | 32 +++-- frontend/src/styles/screens/live-run.css | 9 ++ 25 files changed, 761 insertions(+), 56 deletions(-) create mode 100644 control-plane/tests/test_sandbox_image.py create mode 100644 frontend/src/screens/live/run/RetryTask.tsx diff --git a/control-plane/mergit_cp/config.py b/control-plane/mergit_cp/config.py index ee636a89..cfd2afdd 100644 --- a/control-plane/mergit_cp/config.py +++ b/control-plane/mergit_cp/config.py @@ -50,6 +50,8 @@ class Settings(BaseSettings): """Ask GitHub whether the repository and branch exist (publicly) before a plan is drafted.""" max_attempts: int = Field(default=3, ge=1) """Agents a requester may list per task: the pick plus the retries (plan O6: at most 3 attempts).""" + max_requester_attempts: int = Field(default=6, ge=1) + """Attempts one task may reach in all, counting those the requester starts by hand once it is stuck.""" events_token: SecretStr """Shared secret sent to the gateway as ``Authorization: Bearer`` (spec/events-v0.md).""" diff --git a/control-plane/mergit_cp/dispatch.py b/control-plane/mergit_cp/dispatch.py index 7482c023..1fae56d4 100644 --- a/control-plane/mergit_cp/dispatch.py +++ b/control-plane/mergit_cp/dispatch.py @@ -29,6 +29,11 @@ DISPATCH_DEADLINE_S = 1800 """How long a dispatched task has to be sealed before the plan may give up on it (spec/events-v0.md).""" +RETRY_WINDOW_S = DISPATCH_DEADLINE_S + 600 +"""How long a retry's dispatch keeps being delivered. The gateway refuses to lock attempt n while attempt n-1 +still holds its payment (503), and an attempt that ended without a receipt only gets it back once its deadline +passes and the keeper returns it. A shorter window gives the retry up before then, for good.""" + async def _task_states(conn: AsyncConnection[TupleRow], plan_uuid: str) -> list[TaskState]: cur = await conn.execute( @@ -85,7 +90,11 @@ async def dispatch_task( "deadline_unix": int(time.time()) + DISPATCH_DEADLINE_S, }, plan_id=plan.plan_id, - retry_window_s=settings.event_retry_window_s, + retry_window_s=( + settings.event_retry_window_s + if attempt == 1 + else max(settings.event_retry_window_s, RETRY_WINDOW_S) + ), ) return task_id diff --git a/control-plane/mergit_cp/events.py b/control-plane/mergit_cp/events.py index a71ce14c..b071eeff 100644 --- a/control-plane/mergit_cp/events.py +++ b/control-plane/mergit_cp/events.py @@ -5,7 +5,8 @@ ``Idempotency-Key``. 200 is delivered; another 4xx means the gateway rejected the event for good; 5xx and transport errors are retried with backoff until the event's retry window ends, then it is marked failed and logged as an error for an operator. Rows are claimed with SKIP LOCKED, so several API instances can -deliver without sending one event twice, and an event waits while an earlier event of its plan is pending. +deliver without sending one event twice, and an event waits while an earlier event of its plan is pending, +unless that earlier event is a retry's dispatch. """ import asyncio @@ -58,6 +59,9 @@ async def enqueue( AND NOT EXISTS ( SELECT 1 FROM outbox e WHERE e.plan_id = o.plan_id AND e.id < o.id AND e.delivered_at IS NULL AND e.failed_at IS NULL + -- A retry's dispatch can wait half an hour for its previous attempt's payment + -- (dispatch.RETRY_WINDOW_S), and nothing sent after it depends on it, so it holds up nothing. + AND NOT (e.type = 'task.dispatch_requested' AND (e.body->'payload'->>'attempt')::int > 1) ) ORDER BY id LIMIT %s diff --git a/control-plane/mergit_cp/outcomes.py b/control-plane/mergit_cp/outcomes.py index 629461ef..80d4a2e4 100644 --- a/control-plane/mergit_cp/outcomes.py +++ b/control-plane/mergit_cp/outcomes.py @@ -59,13 +59,29 @@ class FinishedJob: LIMIT %s """ -_PROCESSED_INDEXES = """ -SELECT "index", bool_or(phi_holds = true) AS passed, max(attempt) AS last_attempt -FROM jobs -WHERE plan_uuid = %s AND outcome_processed_at IS NOT NULL -GROUP BY "index" +_DISPATCHES = """ +SELECT d."index", d.attempt, + j.task_id IS NOT NULL, + j.outcome_processed_at IS NOT NULL, + coalesce(j.phi_holds, false), + EXISTS (SELECT 1 FROM outbox o + WHERE o.type = 'task.dispatch_requested' AND o.body->'payload'->>'task_id' = d.task_id + AND o.failed_at IS NOT NULL) +FROM task_dispatches d LEFT JOIN jobs j ON j.task_id = d.task_id +WHERE d.plan_uuid = %s """ + +@dataclass(frozen=True, slots=True) +class _Dispatch: + attempt: int + has_job: bool + processed: bool + holds: bool + given_up: bool + """Its `task.dispatch_requested` was never delivered: the gateway never locked it, so no job will come.""" + + _ALL_PASSED = """ SELECT NOT EXISTS ( SELECT 1 FROM tasks t WHERE t.plan_uuid = %s AND NOT EXISTS ( @@ -150,23 +166,57 @@ async def process_due(conn: AsyncConnection[TupleRow], settings: Settings, limit async def stuck_indexes(conn: AsyncConnection[TupleRow], plan: StoredPlan, settings: Settings) -> list[int]: - """Indexes whose latest processed attempt did not pass and will never be retried: the attempt cap or - the requester's agent list is exhausted. ``needs_requester`` in plan task O6's own words, computed on - demand from ``jobs`` and ``tasks.agents`` rather than stored, since it is exactly what - ``dispatch_ready`` already would not find ready.""" - cur = await conn.execute(_PROCESSED_INDEXES, (plan.plan_uuid,)) - by_index = {index: (passed, last_attempt) for index, passed, last_attempt in await cur.fetchall()} + """Indexes nothing will move on without the requester (``needs_requester`` in plan task O6's own + words): the newest attempt did not pass and no retry follows it (the attempt cap or the requester's + agent list is exhausted), or the newest attempt's dispatch was given up before the gateway locked it. + Computed on demand from ``task_dispatches``, ``jobs``, the outbox and ``tasks.agents`` rather than stored. + An attempt still being delivered, waiting on its escrow, or running is in flight, not stuck.""" + cur = await conn.execute(_DISPATCHES, (plan.plan_uuid,)) + by_index: dict[int, list[_Dispatch]] = {} + for index, attempt, has_job, processed, holds, given_up in await cur.fetchall(): + by_index.setdefault(index, []).append(_Dispatch(attempt, has_job, processed, holds, given_up)) stuck: list[int] = [] for task in plan.tasks: - state = by_index.get(task.draft.index) - if state is None: - continue # never dispatched, or still in flight - passed, last_attempt = state - if not passed and _next_agent(task, int(last_attempt), settings.max_attempts) is None: + sent = by_index.get(task.draft.index) + if not sent or any(d.holds for d in sent): + continue # never dispatched, or passed + newest = max(sent, key=lambda d: d.attempt) + if not newest.has_job: + if newest.given_up: + stuck.append(task.draft.index) + continue + if newest.processed and _next_agent(task, newest.attempt, settings.max_attempts) is None: stuck.append(task.draft.index) return stuck +async def next_attempt(conn: AsyncConnection[TupleRow], plan: StoredPlan, index: int) -> int: + cur = await conn.execute( + 'SELECT coalesce(max(attempt), 0) + 1 FROM task_dispatches WHERE plan_uuid = %s AND "index" = %s', + (plan.plan_uuid, index), + ) + row = await cur.fetchone() + assert row is not None + return int(row[0]) + + +async def retry( + conn: AsyncConnection[TupleRow], plan: StoredPlan, index: int, agent_id: str, settings: Settings +) -> str: + """The requester sends a stuck index to one more agent. The agent takes slot ``attempt`` of + ``tasks.agents`` (any later pick a given-up dispatch never reached is dropped), so attempt ``n`` keeps + naming ``agents[n - 1]`` as everywhere else. It is paid from the index's own escrow, which comes back to + the plan when the failed attempt's money is returned (the gateway holds the dispatch until then). The + caller checks the index is stuck and the agent exists.""" + attempt = await next_attempt(conn, plan, index) + await conn.execute( + "UPDATE tasks SET agents = coalesce(agents[1:%s], ARRAY[]::text[]) || ARRAY[%s::text]" + ' WHERE plan_uuid = %s AND "index" = %s', + (attempt - 1, agent_id, plan.plan_uuid, index), + ) + return await dispatch.dispatch_task(conn, plan, index, attempt, agent_id, settings) + + async def abandon(conn: AsyncConnection[TupleRow], plan: StoredPlan, settings: Settings) -> None: """The requester gives up on this plan. A plan may be abandoned at any point past ``draft`` (not only once every live index is ``needs_requester``: the requester may have other reasons), and this is @@ -176,6 +226,14 @@ async def abandon(conn: AsyncConnection[TupleRow], plan: StoredPlan, settings: S " AND state NOT IN ('closed', 'done')", (plan.plan_uuid,), ) + # An attempt not yet sent to the gateway must not lock money after the requester gave up, and a retry + # waiting on its escrow would otherwise hold `plan.closed` back for up to half an hour. + await conn.execute( + "UPDATE outbox SET failed_at = now(), last_error = 'the requester closed the plan first'" + " WHERE plan_id = %s AND type = 'task.dispatch_requested'" + " AND delivered_at IS NULL AND failed_at IS NULL", + (plan.plan_id,), + ) await enqueue( conn, "plan.closed", diff --git a/control-plane/mergit_cp/routes_plans.py b/control-plane/mergit_cp/routes_plans.py index 8062f739..88b37132 100644 --- a/control-plane/mergit_cp/routes_plans.py +++ b/control-plane/mergit_cp/routes_plans.py @@ -68,6 +68,12 @@ class ApprovePlan(BaseModel): funding: Literal["wallet", "credits"] +class RetryTask(BaseModel): + model_config = ConfigDict(extra="forbid") + + agent_id: str + + def _problems(status: int, problems: list[str]) -> HTTPException: return HTTPException(status_code=status, detail={"problems": problems}) @@ -337,3 +343,39 @@ async def abandon_plan( raise HTTPException(status_code=409, detail=f"the plan is already {plan.state}") await outcomes.abandon(conn, plan, settings) return {"plan_id": plan.plan_id, "state": "closed"} + + +@router.post("/{plan_uuid}/tasks/{index}/retry") +async def retry_task( + plan_uuid: UUID, + index: int, + body: RetryTask, + principal: PrincipalDep, + settings: SettingsDep, + pool: PoolDep, + agents: Annotated[list[Agent], Depends(get_agents)], +) -> dict[str, Any]: + """The requester sends a task that waits on them (every picked agent failed, or its dispatch was given + up) to one more agent of their choice, instead of closing the whole plan.""" + async with pool.connection() as conn, conn.transaction(): + plan = await plans.load(conn, plan_uuid, for_update=True) + if plan is None: + raise HTTPException(status_code=404, detail="no such plan") + if not principal.owns(plan.requester): + raise HTTPException(status_code=403, detail="only the requester may retry a task of this plan") + if plan.state != "running": + raise HTTPException(status_code=409, detail=f"the plan is {plan.state}, not running") + if all(t.draft.index != index for t in plan.tasks): + raise HTTPException(status_code=404, detail=f"the plan has no task {index}") + if all(a.agent_id != body.agent_id for a in agents): + raise _problems(422, [f"agent {body.agent_id} is not deployed"]) + if index not in await outcomes.stuck_indexes(conn, plan, settings): + raise HTTPException(status_code=409, detail=f"task {index} is not waiting on you") + attempt = await outcomes.next_attempt(conn, plan, index) + if attempt > settings.max_requester_attempts: + raise HTTPException( + status_code=409, + detail=f"task {index} has had {attempt - 1} attempts, the most one task may have", + ) + task_id = await outcomes.retry(conn, plan, index, body.agent_id, settings) + return {"task_id": task_id, "index": index, "attempt": attempt, "agent_id": body.agent_id} diff --git a/control-plane/mergit_cp/routes_runs.py b/control-plane/mergit_cp/routes_runs.py index 80d047a4..df5d7d91 100644 --- a/control-plane/mergit_cp/routes_runs.py +++ b/control-plane/mergit_cp/routes_runs.py @@ -64,14 +64,20 @@ async def list_plans(principal: PrincipalDep, pool: PoolDep, settings: SettingsD @router.get("/{plan_uuid}") -async def plan_detail(plan_uuid: UUID, principal: PrincipalDep, pool: PoolDep) -> dict[str, Any]: +async def plan_detail( + plan_uuid: UUID, principal: PrincipalDep, pool: PoolDep, settings: SettingsDep +) -> dict[str, Any]: async with pool.connection() as conn: plan = await _owned_plan(conn, plan_uuid, principal) attempts = await runs.attempts(conn, plan_uuid) ledger = await money.list_for_plan(conn, plan_uuid) + pending = await runs.pending_dispatches(conn, plan_uuid) + stuck = await outcomes.stuck_indexes(conn, plan, settings) if plan.state == "running" else [] view = plans.view(plan) view["attempts"] = [runs.attempt_view(a) for a in attempts] view["money_events"] = [money.view(m) for m in ledger] + view["pending_dispatches"] = [runs.pending_view(p) for p in pending] + view["stuck_indexes"] = stuck return view diff --git a/control-plane/mergit_cp/runs.py b/control-plane/mergit_cp/runs.py index e2d93e47..307043d1 100644 --- a/control-plane/mergit_cp/runs.py +++ b/control-plane/mergit_cp/runs.py @@ -86,6 +86,57 @@ async def attempts(conn: AsyncConnection[TupleRow], plan_uuid: UUID | str) -> li ] +@dataclass(frozen=True, slots=True) +class PendingDispatch: + """An attempt the orchestrator asked for that has no job yet: still being sent to the gateway (a retry + waits there until the previous attempt's payment is back), sent and waiting for its escrow to lock, or + given up.""" + + index: int + attempt: int + agent_id: str + state: str + last_error: str | None + give_up_at: datetime | None + + +_PENDING = """ +SELECT d."index", d.attempt, d.agent_id, o.delivered_at IS NOT NULL, o.failed_at IS NOT NULL, o.last_error, + o.give_up_at +FROM task_dispatches d +LEFT JOIN jobs j ON j.task_id = d.task_id +LEFT JOIN outbox o ON o.type = 'task.dispatch_requested' AND o.body->'payload'->>'task_id' = d.task_id +WHERE d.plan_uuid = %s AND j.task_id IS NULL +ORDER BY d."index", d.attempt +""" + + +async def pending_dispatches(conn: AsyncConnection[TupleRow], plan_uuid: UUID | str) -> list[PendingDispatch]: + cur = await conn.execute(_PENDING, (str(plan_uuid),)) + return [ + PendingDispatch( + index=index, + attempt=attempt, + agent_id=agent_id, + state="given_up" if failed else "escrowing" if delivered else "sending", + last_error=last_error, + give_up_at=give_up_at, + ) + for index, attempt, agent_id, delivered, failed, last_error, give_up_at in await cur.fetchall() + ] + + +def pending_view(p: PendingDispatch) -> dict[str, Any]: + return { + "index": p.index, + "attempt": p.attempt, + "agent_id": p.agent_id, + "state": p.state, + "last_error": p.last_error, + "give_up_at": None if p.give_up_at is None else p.give_up_at.isoformat(), + } + + def attempt_view(a: Attempt) -> dict[str, Any]: return { "task_id": a.task_id, diff --git a/control-plane/mergit_worker/__main__.py b/control-plane/mergit_worker/__main__.py index 43d28613..6a3d4409 100644 --- a/control-plane/mergit_worker/__main__.py +++ b/control-plane/mergit_worker/__main__.py @@ -8,6 +8,8 @@ import threading from dataclasses import dataclass +import docker.errors + from mergit_cp.pr_store import PgPrStore from .config import ConfigError, WorkerConfig, load_config @@ -19,7 +21,7 @@ from .preflight import check_models from .progress import HttpProgressReporter, NullProgressReporter, ProgressReporter from .publish import ReceiptPublisher -from .sandbox import Sandbox, SandboxConfig, reap_stale +from .sandbox import Sandbox, SandboxConfig, SandboxError, reap_stale, require_image from .startup import WorkerIdLock, agents_without_key, sweep_workspaces from .tools import ToolLimits from .worker import Worker, WorkerDeps @@ -161,6 +163,11 @@ def main() -> int: log.warning( "no usable key for agent(s) %s: a job for them fails, and the requester loses an attempt", names ) + try: + require_image(cfg.sandbox_image) + except (SandboxError, docker.errors.DockerException) as exc: + log.error("cannot start sandboxes, so no job is claimed: %s", exc) + return 2 lock = WorkerIdLock.acquire(cfg.database_url.get_secret_value(), worker_id) if lock is None: log.error( diff --git a/control-plane/mergit_worker/sandbox.py b/control-plane/mergit_worker/sandbox.py index 0650e97b..3c06c2b5 100644 --- a/control-plane/mergit_worker/sandbox.py +++ b/control-plane/mergit_worker/sandbox.py @@ -49,6 +49,19 @@ class SandboxError(Exception): """The sandbox could not be created or used.""" +def require_image(image: str, client: docker.DockerClient | None = None) -> None: + """The sandbox image is on this host. Checked at start-up too: a worker without it would fail every job + it claims, each costing the requester an agent pick, so it refuses to start and jobs stay queued.""" + dc = client or docker.from_env() + try: + dc.images.get(image) + except docker.errors.ImageNotFound as exc: + raise SandboxError( + f"sandbox image {image!r} is not built on this host:" + f" docker build -t {image} control-plane/sandbox" + ) from exc + + class SandboxPort(Protocol): """What the harness needs from a sandbox. `Sandbox` is the production implementation; tests that do not need isolation provide a process-based stand-in.""" @@ -98,10 +111,7 @@ def create( client: docker.DockerClient | None = None, ) -> Sandbox: dc = client or docker.from_env() - try: - dc.images.get(cfg.image) - except docker.errors.ImageNotFound as exc: - raise SandboxError(f"sandbox image {cfg.image!r} is not built on this host") from exc + require_image(cfg.image, dc) workspace = Workspace.clone(repo_url, ref, cfg.workspace_root, cfg.git) try: container = _create_container(dc, workspace.work, cfg) diff --git a/control-plane/tests/test_api.py b/control-plane/tests/test_api.py index 2959301e..91299483 100644 --- a/control-plane/tests/test_api.py +++ b/control-plane/tests/test_api.py @@ -179,6 +179,51 @@ async def every_pick_failed(pool: Pool) -> None: assert body[moving_plan]["stuck_tasks"] == 0 +def test_a_stuck_task_can_be_retried_with_another_agent(client: TestClient, settings: Settings): + address = sign_in(client) + plan_uuid, _, task_ids = run(settings.database_url, lambda pool: seed_plan(pool, requester=address)) + + async def every_pick_failed(pool: Pool) -> None: + async with pool.connection() as conn: + await conn.execute( + "UPDATE jobs SET status = 'error', finished_at = now(), outcome_processed_at = now()" + " WHERE task_id = %s", + (task_ids[1],), + ) + + run(settings.database_url, every_pick_failed) + before = client.get(f"/v0/plans/{plan_uuid}").json() + assert (before["stuck_indexes"], before["pending_dispatches"]) == ([1], []) + + assert client.post(f"/v0/plans/{plan_uuid}/tasks/1/retry", json={"agent_id": "9"}).status_code == 422 + assert client.post(f"/v0/plans/{plan_uuid}/tasks/0/retry", json={"agent_id": "1"}).status_code == 409 + assert client.post(f"/v0/plans/{plan_uuid}/tasks/7/retry", json={"agent_id": "1"}).status_code == 404 + + r = client.post(f"/v0/plans/{plan_uuid}/tasks/1/retry", json={"agent_id": "1"}) + assert r.status_code == 200, r.text + assert (r.json()["attempt"], r.json()["agent_id"]) == (2, "1") + + after = client.get(f"/v0/plans/{plan_uuid}").json() + assert after["stuck_indexes"] == [] + assert after["tasks"][1]["agents"] == ["1", "1"] + [pending] = after["pending_dispatches"] + assert (pending["index"], pending["attempt"], pending["agent_id"], pending["state"]) == ( + 1, + 2, + "1", + "sending", + ) + # Retrying again before that attempt has run is refused: the task no longer waits on the requester. + assert client.post(f"/v0/plans/{plan_uuid}/tasks/1/retry", json={"agent_id": "1"}).status_code == 409 + + +def test_only_the_requester_may_retry_a_task(client: TestClient, settings: Settings): + other = Account.create().address.lower() + plan_uuid, _, _ = run(settings.database_url, lambda pool: seed_plan(pool, requester=other)) + sign_in(client) + assert client.post(f"/v0/plans/{plan_uuid}/tasks/1/retry", json={"agent_id": "1"}).status_code == 403 + + def test_plans_list_requires_a_session(client: TestClient): assert client.get("/v0/plans").status_code == 401 diff --git a/control-plane/tests/test_events.py b/control-plane/tests/test_events.py index 33c229f5..6bd085b6 100644 --- a/control-plane/tests/test_events.py +++ b/control-plane/tests/test_events.py @@ -189,6 +189,59 @@ async def body(pool: Pool) -> None: assert all(r["delivered"] for r in rows(database_url)) +def dispatch(attempt: int, plan_id: str = PLAN_A) -> dict[str, Any]: + return { + "task_id": "0x" + f"{attempt:x}" * 64, + "plan_id": plan_id, + "index": 0, + "attempt": attempt, + "agent_id": "1", + "amount_usdc_units": "500000", + "category": "test-repair", + "phi_hash": "0x" + "c" * 64, + "deadline_unix": 2_000_000_000, + } + + +def test_a_retry_waiting_on_its_escrow_holds_up_no_other_event_of_its_plan(database_url: str) -> None: + sent: list[str] = [] + + async def gateway(request: httpx.Request) -> httpx.Response: + envelope = json.loads(request.content) + sent.append(f"{envelope['type']}:{envelope['payload'].get('attempt', '')}") + # The gateway holds a retry while the previous attempt still holds its payment. + return httpx.Response(503 if envelope["payload"].get("attempt") == 2 else 200) + + async def body(pool: Pool) -> None: + async with pool.connection() as conn: + await enqueue(conn, "task.dispatch_requested", dispatch(2), plan_id=PLAN_A, retry_window_s=2400) + await enqueue(conn, "task.dispatch_requested", dispatch(1), plan_id=PLAN_A, retry_window_s=600) + await deliverer(pool, gateway).deliver_due() + + run(database_url, body) + assert sent == ["task.dispatch_requested:2", "task.dispatch_requested:1"] + retry, other = rows(database_url) + assert (retry["delivered"], retry["failed"], retry["deferred"]) == (False, False, True) + assert other["delivered"] + + +def test_a_first_attempt_still_holds_up_the_events_after_it(database_url: str) -> None: + sent: list[int] = [] + + async def down_once(request: httpx.Request) -> httpx.Response: + sent.append(json.loads(request.content)["payload"]["attempt"]) + return httpx.Response(503) + + async def body(pool: Pool) -> None: + async with pool.connection() as conn: + await enqueue(conn, "task.dispatch_requested", dispatch(1), plan_id=PLAN_A, retry_window_s=600) + await enqueue(conn, "task.dispatch_requested", dispatch(2), plan_id=PLAN_A, retry_window_s=2400) + await deliverer(pool, down_once).deliver_due() + + run(database_url, body) + assert sent == [1] + + def test_two_deliverers_never_send_the_same_event_twice(database_url: str) -> None: sent: list[str] = [] diff --git a/control-plane/tests/test_outcomes.py b/control-plane/tests/test_outcomes.py index a89fc71c..b215c825 100644 --- a/control-plane/tests/test_outcomes.py +++ b/control-plane/tests/test_outcomes.py @@ -389,3 +389,136 @@ async def body(pool: Pool) -> None: assert n is not None and n[0] == 2 # both events still sent; the state change was the idempotent part run(database_url, body) + + +async def _fail(conn: psycopg.AsyncConnection[TupleRow], task_id: str) -> None: + await conn.execute( + "UPDATE jobs SET status = 'error', phi_holds = NULL, finished_at = now() WHERE task_id = %s", + (task_id,), + ) + + +async def _window_s(conn: psycopg.AsyncConnection[TupleRow], attempt: int) -> float: + cur = await conn.execute( + "SELECT extract(epoch FROM give_up_at - created_at) FROM outbox" + " WHERE type = 'task.dispatch_requested' AND (body->'payload'->>'attempt')::int = %s", + (attempt,), + ) + row = await cur.fetchone() + assert row is not None + return float(row[0]) + + +def test_a_retry_outlasts_the_previous_attempts_deadline(database_url: str) -> None: + """The gateway holds attempt 2 until attempt 1's payment is back, which for an attempt that never + sealed a receipt is its deadline. A retry given up before then never runs.""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + + async with pool.connection() as conn: + first, retry = await _window_s(conn, 1), await _window_s(conn, 2) + stuck = await outcomes.stuck_indexes(conn, plan, SETTINGS) + assert first == pytest.approx(SETTINGS.event_retry_window_s, abs=5) + assert retry > outcomes.dispatch.DISPATCH_DEADLINE_S + 60 + assert stuck == [] # on its way, not waiting on the requester + + run(database_url, body) + + +def test_a_retry_given_up_before_it_started_waits_on_the_requester(database_url: str) -> None: + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + async with pool.connection() as conn: + await conn.execute( + "UPDATE outbox SET failed_at = now(), last_error = 'gave up'" + " WHERE (body->'payload'->>'attempt')::int = 2" + ) + stuck = await outcomes.stuck_indexes(conn, plan, SETTINGS) + assert stuck == [0] + + run(database_url, body) + + +def test_closing_cancels_attempts_not_yet_sent(database_url: str) -> None: + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + await conn.execute("UPDATE plans SET state = 'running' WHERE plan_uuid = %s", (plan.plan_uuid,)) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await conn.execute("UPDATE outbox SET delivered_at = now()") # attempt 1 reached the gateway + await _process(pool) # dispatches attempt 2, which the gateway would hold + + async with pool.connection() as conn, conn.transaction(): + loaded = await plans.load(conn, uuid.UUID(plan.plan_uuid), for_update=True) + assert loaded is not None + await outcomes.abandon(conn, loaded, SETTINGS) + + async with pool.connection() as conn: + rows = await ( + await conn.execute( + "SELECT type, delivered_at IS NOT NULL, failed_at IS NOT NULL, last_error" + " FROM outbox ORDER BY id" + ) + ).fetchall() + assert rows == [ + ("task.dispatch_requested", True, False, None), + ("task.dispatch_requested", False, True, "the requester closed the plan first"), + ("plan.closed", False, False, None), + ] + + run(database_url, body) + + +def test_a_requester_retry_takes_the_next_slot_and_dispatches(database_url: str) -> None: + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + + async with pool.connection() as conn, conn.transaction(): + assert await outcomes.stuck_indexes(conn, plan, SETTINGS) == [0] + task_id = await outcomes.retry(conn, plan, 0, "3", SETTINGS) + + async with pool.connection() as conn: + loaded = await plans.load(conn, uuid.UUID(plan.plan_uuid)) + dispatched = await ( + await conn.execute( + "SELECT attempt, agent_id FROM task_dispatches WHERE task_id = %s", (task_id,) + ) + ).fetchone() + stuck = await outcomes.stuck_indexes(conn, plan, SETTINGS) + window = await _window_s(conn, 2) + assert loaded is not None and loaded.tasks[0].agents == ["1", "3"] + assert dispatched == (2, "3") + assert stuck == [] + assert window > outcomes.dispatch.DISPATCH_DEADLINE_S + + run(database_url, body) + + +def test_a_retry_after_a_given_up_dispatch_reuses_its_slot(database_url: str) -> None: + """Attempt n always names agents[n - 1]: a pick the given-up attempt never reached is replaced.""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2", "3"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + async with pool.connection() as conn, conn.transaction(): + await conn.execute( + "UPDATE outbox SET failed_at = now() WHERE (body->'payload'->>'attempt')::int = 2" + ) + assert await outcomes.stuck_indexes(conn, plan, SETTINGS) == [0] + await outcomes.retry(conn, plan, 0, "1", SETTINGS) + loaded = await plans.load(conn, uuid.UUID(plan.plan_uuid)) + assert loaded is not None and loaded.tasks[0].agents == ["1", "2", "1"] + + run(database_url, body) diff --git a/control-plane/tests/test_sandbox_image.py b/control-plane/tests/test_sandbox_image.py new file mode 100644 index 00000000..253510b0 --- /dev/null +++ b/control-plane/tests/test_sandbox_image.py @@ -0,0 +1,36 @@ +"""The worker refuses to start without its sandbox image, rather than failing every job it would claim.""" + +from __future__ import annotations + +from typing import Any, cast + +import docker +import docker.errors +import pytest + +from mergit_worker.sandbox import SandboxError, require_image + + +class _Images: + def __init__(self, present: set[str]) -> None: + self._present = present + + def get(self, name: str) -> object: + if name not in self._present: + raise docker.errors.ImageNotFound(name) + return object() + + +def _client(*present: str) -> docker.DockerClient: + fake: Any = type("FakeDocker", (), {})() + fake.images = _Images(set(present)) + return cast(docker.DockerClient, fake) + + +def test_a_missing_image_says_how_to_build_it() -> None: + with pytest.raises(SandboxError, match=r"docker build -t mergit-sandbox:dev control-plane/sandbox"): + require_image("mergit-sandbox:dev", _client()) + + +def test_a_built_image_passes() -> None: + require_image("mergit-sandbox:dev", _client("mergit-sandbox:dev")) diff --git a/frontend/e2e/live/plans.ts b/frontend/e2e/live/plans.ts index 734d4ddf..3ebec488 100644 --- a/frontend/e2e/live/plans.ts +++ b/frontend/e2e/live/plans.ts @@ -52,6 +52,14 @@ export type FakePlan = { tasks: FakeTask[]; attempts: FakeAttempt[]; money_events: Array<{ type: string; at: string; payload: Record }>; + pending_dispatches?: Array<{ + index: number; + attempt: number; + agent_id: string; + state: "sending" | "escrowing" | "given_up"; + last_error: string | null; + give_up_at: string | null; + }>; }; /** One Server-Sent Event the fake stream sends, as the control plane names them. */ @@ -319,6 +327,28 @@ export async function mockPlanApi( if (prState === "queued" || prState === "running") prState = prPolls.shift() ?? "done"; return json(route, prBody()); } + if (method === "POST" && action === "tasks" && path.endsWith("/retry")) { + // POST /v0/plans/{uuid}/tasks/{index}/retry: the next attempt goes to the gateway, which holds it until + // the failed attempt's payment is back. + const index = Number(path.split("/")[5]); + const { agent_id } = body as { agent_id: string }; + const task = plan.tasks.find((t) => t.index === index); + if (!task) return json(route, { detail: `the plan has no task ${String(index)}` }, 404); + const attempt = Math.max(0, ...plan.attempts.filter((a) => a.index === index).map((a) => a.attempt)) + 1; + task.agents = [...(task.agents ?? []).slice(0, attempt - 1), agent_id]; + plan.pending_dispatches = [ + ...(plan.pending_dispatches ?? []), + { + index, + attempt, + agent_id, + state: "sending", + last_error: `503: attempt ${String(attempt - 1)} of this task still holds its payment`, + give_up_at: null, + }, + ]; + return json(route, { task_id: taskId(index, attempt), index, attempt, agent_id }); + } if (method === "POST" && action === "abandon") { plan.state = "closed"; return json(route, { plan_id: plan.plan_id, state: "closed" }); diff --git a/frontend/e2e/live/run.spec.ts b/frontend/e2e/live/run.spec.ts index e10b9cc5..fc95087d 100644 --- a/frontend/e2e/live/run.spec.ts +++ b/frontend/e2e/live/run.spec.ts @@ -227,7 +227,7 @@ test("an inconclusive task with no agent left asks the requester, who can close await page.goto(`/app/runs/${stuck.plan_uuid}`); const banner = page.getByRole("alert"); - await expect(banner).toContainText("Needs you: task 1 failed its check with every agent you picked."); + await expect(banner).toContainText("Needs you: task 1 cannot go on without you."); await expect(page.getByRole("complementary", { name: "Task 1 details" })).toContainText("Inconclusive"); await banner.getByRole("button", { name: "Close plan" }).click(); @@ -236,6 +236,33 @@ test("an inconclusive task with no agent left asks the requester, who can close expect(only(calls, "POST", "/abandon")).toHaveLength(1); }); +test("a stuck task can be sent to another agent instead of closing the plan", async ({ page }) => { + await mockBackend(page); + const stuck = approvedPlan( + { + state: "running", + attempts: [attempt(0, 1, { status: "error", phi_holds: null, agent_id: "1", notes: "SandboxError" })], + }, + ["1"], + ); + const calls = await mockPlanApi(page, { plans: [stuck] }); + await page.goto(`/app/runs/${stuck.plan_uuid}`); + + const banner = page.getByRole("alert"); + const picker = banner.getByRole("combobox", { name: "Agent to retry task 1 with" }); + // An agent this task has not had yet is offered first. + await expect(picker).toHaveValue("2"); + await banner.getByRole("button", { name: "Retry with this agent" }).click(); + + await expect(page.getByRole("alert")).toHaveCount(0); + expect(only(calls, "POST", "/retry").map((c) => [c.path, c.body])).toEqual([ + [`/v0/plans/${stuck.plan_uuid}/tasks/0/retry`, { agent_id: "2" }], + ]); + const panel = page.getByRole("complementary", { name: "Task 1 details" }); + await expect(panel).toContainText("Retrying"); + await expect(panel).toContainText("starts once attempt 1's payment is back in the plan"); +}); + test("the runs tab lists the requester's plans and opens one", async ({ page }) => { await mockBackend(page); await mockPlanApi(page, { plans: [draftPlan()] }); diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index a20a9fca..9daae9e0 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -111,6 +111,14 @@ export const approvePlan = (planUuid: string, body: ApproveRequest) => export const abandonPlan = (planUuid: string) => request<{ plan_id: string; state: "closed" }>("POST", `${planPath(planUuid)}/abandon`); +/** Sends a task that waits on the requester to one more agent; it is paid from that task's own escrow. */ +export const retryTask = (planUuid: string, index: number, agentId: string) => + request<{ task_id: string; index: number; attempt: number; agent_id: string }>( + "POST", + `${planPath(planUuid)}/tasks/${String(index)}/retry`, + { agent_id: agentId }, + ); + export const listPlans = () => request("GET", "/v0/plans"); export const getPlan = (planUuid: string) => request("GET", planPath(planUuid)); diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index cf18ad03..5cd3acb5 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -60,7 +60,25 @@ export type Attempt = { /** A gateway event about this plan's money or verdicts, as the gateway sent it. */ export type MoneyEvent = { type: string; at: string; payload: Record }; -export type PlanDetail = Plan & { attempts: Attempt[]; money_events: MoneyEvent[] }; +/** + * An attempt the orchestrator asked for that has no job yet. `sending`: still being delivered to the gateway, + * which holds a retry until the previous attempt's payment is back (`last_error` says so). `escrowing`: sent, + * its payment not locked yet. `given_up`: never started. + */ +export type PendingDispatch = { + index: number; + attempt: number; + agent_id: string; + state: "sending" | "escrowing" | "given_up"; + last_error: string | null; + give_up_at: string | null; +}; + +export type PlanDetail = Plan & { + attempts: Attempt[]; + money_events: MoneyEvent[]; + pending_dispatches?: PendingDispatch[]; +}; export type PlanSummary = Pick< Plan, diff --git a/frontend/src/screens/live/run/LiveRunWorkspace.tsx b/frontend/src/screens/live/run/LiveRunWorkspace.tsx index 496c878b..0e8012e7 100644 --- a/frontend/src/screens/live/run/LiveRunWorkspace.tsx +++ b/frontend/src/screens/live/run/LiveRunWorkspace.tsx @@ -1,3 +1,4 @@ +import { useEffect } from "react"; import { useParams } from "react-router-dom"; import { paths } from "../../../app/paths"; import { useAgents } from "../../../api/useAgents"; @@ -6,13 +7,16 @@ import type { AgentInfo, PlanDetail, StepEvent } from "../../../api/types"; import { MergitLoader } from "../../../components/MergitLoader"; import { PageHeader } from "../../../components/PageHeader"; import { PlanBadge } from "./badges"; -import { stuckTasks } from "./progress"; +import { stuckTasks, taskProgress } from "./progress"; import { DraftPlanEditor } from "./DraftPlanEditor"; import { FundingStep } from "./FundingStep"; import { RunView } from "./RunView"; import "../../../styles/screens/run.css"; import "../../../styles/screens/live-run.css"; +/** The stream carries jobs, not dispatches, so a retry on its way to the gateway is followed by reloading. */ +const RELOAD_WHILE_RETRYING_MS = 15_000; + const STREAM_LABEL: Record = { connecting: "Connecting…", open: "Live", @@ -27,6 +31,16 @@ export function LiveRunWorkspace() { const { runId = "" } = useParams(); const { detail, steps, error, stream, reload } = usePlan(runId); const { agents } = useAgents(); + const retrying = + detail?.state === "running" && + detail.tasks.some((t) => taskProgress(t, detail.attempts, detail.pending_dispatches ?? []).kind === "retrying"); + useEffect(() => { + if (!retrying) return; + const timer = setInterval(reload, RELOAD_WHILE_RETRYING_MS); + return () => { + clearInterval(timer); + }; + }, [retrying, reload]); if (!detail) { return ( @@ -50,7 +64,12 @@ export function LiveRunWorkspace() { {detail.goal}} crumb={{ label: "Runs", to: paths.runs, id: detail.plan_uuid.slice(0, 8) }} - badge={} + badge={ + + } subtitle={ <> {detail.repo_url} at {detail.ref} diff --git a/frontend/src/screens/live/run/RetryTask.tsx b/frontend/src/screens/live/run/RetryTask.tsx new file mode 100644 index 00000000..9572e8ea --- /dev/null +++ b/frontend/src/screens/live/run/RetryTask.tsx @@ -0,0 +1,64 @@ +import { useState } from "react"; +import { retryTask } from "../../../api/client"; +import type { AgentInfo, PlanTask } from "../../../api/types"; +import { Problems } from "./Problems"; +import { useSubmit } from "./useSubmit"; + +/** + * Sends one stuck task to an agent the requester chooses, instead of closing the whole plan. Offers an agent + * this task has not had yet first; any deployed agent may be chosen. + */ +export function RetryTask({ + planUuid, + task, + agents, + onRetried, +}: { + planUuid: string; + task: PlanTask; + agents: AgentInfo[] | null; + onRetried: () => void; +}) { + const tried = new Set(task.agents ?? []); + const fresh = agents?.find((a) => !tried.has(a.agent_id)) ?? agents?.[0]; + const [agentId, setAgentId] = useState(fresh?.agent_id ?? ""); + const { busy, problems, submit } = useSubmit(); + const chosen = agentId || fresh?.agent_id || ""; + + const retry = () => { + void submit(async () => { + await retryTask(planUuid, task.index, chosen); + onRetried(); + }); + }; + + return ( +
+ + Task {task.index + 1}: {task.title} + +
+ + +
+ +
+ ); +} diff --git a/frontend/src/screens/live/run/RunView.tsx b/frontend/src/screens/live/run/RunView.tsx index ef40f154..426976fa 100644 --- a/frontend/src/screens/live/run/RunView.tsx +++ b/frontend/src/screens/live/run/RunView.tsx @@ -7,6 +7,7 @@ import { Icon } from "../../../components/Icon"; import { ProgressBadge } from "./badges"; import { ClosePlan } from "./ClosePlan"; import { PullRequestStep } from "./PullRequestStep"; +import { RetryTask } from "./RetryTask"; import { attemptsOf, closedBeforeStart, stuckTasks, taskProgress } from "./progress"; import { canStartAgain } from "./startAgain"; import { StepLog } from "./StepLog"; @@ -27,11 +28,13 @@ export function RunView({ }) { const [selected, setSelected] = useState(detail.tasks[0]?.index ?? 0); const task = detail.tasks.find((t) => t.index === selected); - const passed = detail.tasks.filter((t) => taskProgress(t, detail.attempts).kind === "passed").length; + const passed = detail.tasks.filter( + (t) => taskProgress(t, detail.attempts, detail.pending_dispatches ?? []).kind === "passed", + ).length; return ( <> - + {detail.state === "done" && }
@@ -73,6 +76,7 @@ export function RunView({ key={task.index} task={task} attempts={detail.attempts} + pending={detail.pending_dispatches ?? []} money={detail.money_events} steps={steps} agents={agents} @@ -96,7 +100,7 @@ function TaskRow({ selected: boolean; onSelect: () => void; }) { - const progress = taskProgress(task, detail.attempts); + const progress = taskProgress(task, detail.attempts, detail.pending_dispatches ?? []); const latest = attemptsOf(task.index, detail.attempts).at(-1); const agentId = latest?.agent_id ?? task.agents?.[0]; return ( @@ -115,7 +119,15 @@ function TaskRow({ } /** What the requester needs to know about the whole plan right now, if anything. */ -function RunBanner({ detail, onClosed }: { detail: PlanDetail; onClosed: () => void }) { +function RunBanner({ + detail, + agents, + onClosed, +}: { + detail: PlanDetail; + agents: AgentInfo[] | null; + onClosed: () => void; +}) { if (detail.state === "done") { return (

@@ -152,18 +164,21 @@ function RunBanner({ detail, onClosed }: { detail: PlanDetail; onClosed: () => v

); } - const stuck = stuckTasks(detail.tasks, detail.attempts); + const stuck = stuckTasks(detail.tasks, detail.attempts, detail.pending_dispatches ?? []); if (stuck.length === 0) return null; const names = stuck.map((t) => `task ${String(t.index + 1)}`).join(", "); return ( -
+
- Needs you: {names} failed its check with every agent you - picked. + Needs you: {names} cannot go on without you. - Nothing more runs on its own. Close the plan to stop it and get its unspent funds back. + Nothing more runs on its own. Retry a task with another agent (it is paid from that task's own budget), or + close the plan to stop it and get its unspent funds back. + {stuck.map((t) => ( + + ))}
); diff --git a/frontend/src/screens/live/run/TaskDetail.tsx b/frontend/src/screens/live/run/TaskDetail.tsx index 4fac42ea..92591092 100644 --- a/frontend/src/screens/live/run/TaskDetail.tsx +++ b/frontend/src/screens/live/run/TaskDetail.tsx @@ -2,7 +2,7 @@ import { Link } from "react-router-dom"; import { paths } from "../../../app/paths"; import { agentName } from "../../../api/useAgents"; import { EnsIdentity } from "../../../components/EnsIdentity"; -import type { AgentInfo, Attempt, MoneyEvent, PlanTask, StepEvent } from "../../../api/types"; +import type { AgentInfo, Attempt, MoneyEvent, PendingDispatch, PlanTask, StepEvent } from "../../../api/types"; import { formatUsdc } from "../../../chain/funding"; import { OutcomeBadge, ProgressBadge, VerdictBadge } from "./badges"; import { phiCommand } from "./format"; @@ -16,10 +16,20 @@ function progressNote(progress: TaskProgress, agents: AgentInfo[] | null): strin return `${agentName(agents, progress.attempt.agent_id)} is working on attempt ${String(progress.attempt.attempt)}.`; case "passed": return "Its check held. The task is done."; - case "retrying": - return `Attempt ${String(progress.failed.attempt)} did not pass its check. Retrying with ${agentName(agents, progress.nextAgent)}.`; + case "retrying": { + const failed = `Attempt ${String(progress.failed.attempt)} did not pass its check.`; + const next = agentName(agents, progress.nextAgent); + if (progress.pending?.state === "escrowing") return `${failed} Locking the payment for ${next}'s attempt.`; + if (progress.pending?.last_error) + // The gateway holds a retry until the failed attempt's payment is back in the plan. An attempt that + // ended without a receipt only gets it back at its deadline, 30 minutes after it was dispatched. + return `${failed} ${next} starts once attempt ${String(progress.failed.attempt)}'s payment is back in the plan, up to 30 minutes after that attempt began.`; + return `${failed} Retrying with ${next}.`; + } case "stuck": - return "Every agent you picked has tried this task and its check never held."; + return progress.givenUp + ? `Attempt ${String(progress.givenUp.attempt)} could not be started${progress.givenUp.last_error ? `: ${progress.givenUp.last_error}` : "."}` + : "Every agent you picked has tried this task and its check never held."; } } @@ -46,17 +56,19 @@ function AttemptMeta({ steps, verdict, receipt }: { steps: number; verdict: Verd export function TaskDetail({ task, attempts, + pending, money, steps, agents, }: { task: PlanTask; attempts: Attempt[]; + pending: PendingDispatch[]; money: MoneyEvent[]; steps: StepEvent[]; agents: AgentInfo[] | null; }) { - const progress = taskProgress(task, attempts); + const progress = taskProgress(task, attempts, pending); const mine = attemptsOf(task.index, attempts); const stepCount = (taskId: string) => steps.filter((s) => s.task_id === taskId).length; diff --git a/frontend/src/screens/live/run/TaskFlow.tsx b/frontend/src/screens/live/run/TaskFlow.tsx index 3ef6bb55..a0b97629 100644 --- a/frontend/src/screens/live/run/TaskFlow.tsx +++ b/frontend/src/screens/live/run/TaskFlow.tsx @@ -70,7 +70,9 @@ export function TaskFlow({ const width = Math.max(measured, widest * (NODE_W + GAP_X) - GAP_X + 2 * SIDE); const height = TOP * 2 + Math.max(1, rows.length) * ROW_H - (ROW_H - NODE_H); const at = place(rows, width); - const progress = new Map(detail.tasks.map((t) => [t.index, taskProgress(t, detail.attempts)])); + const progress = new Map( + detail.tasks.map((t) => [t.index, taskProgress(t, detail.attempts, detail.pending_dispatches ?? [])]), + ); const draft = detail.state === "draft"; // The graph repeats the task list beside it, so assistive tech gets one summary instead of every node again. diff --git a/frontend/src/screens/live/run/progress.test.ts b/frontend/src/screens/live/run/progress.test.ts index b4f60dc4..7e186de0 100644 --- a/frontend/src/screens/live/run/progress.test.ts +++ b/frontend/src/screens/live/run/progress.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import type { Attempt, MoneyEvent, PlanTask } from "../../../api/types"; +import type { Attempt, MoneyEvent, PendingDispatch, PlanTask } from "../../../api/types"; import { attemptOutcome, closedBeforeStart, creditsDenial, stuckTasks, taskProgress, verdictFor } from "./progress"; const task = (agents: string[] | null = ["1", "2"]): PlanTask => ({ @@ -63,6 +63,43 @@ describe("taskProgress", () => { it("only counts attempts at its own index", () => { expect(taskProgress(task(), [{ ...failed(1), index: 1 }])).toEqual({ kind: "waiting" }); }); + + const sending = (n: number, over: Partial = {}): PendingDispatch => ({ + index: 0, + attempt: n, + agent_id: "3", + state: "sending", + last_error: "503: attempt 1 of this task still holds its payment", + give_up_at: null, + ...over, + }); + + it("names the retry the server actually dispatched, and why it waits", () => { + expect(taskProgress(task(["1", "2"]), [failed(1)], [sending(2)])).toEqual({ + kind: "retrying", + failed: failed(1), + nextAgent: "3", + pending: sending(2), + }); + }); + + it("waits on the requester when the next attempt was given up before it started", () => { + const gaveUp = sending(2, { state: "given_up" }); + expect(taskProgress(task(["1", "2"]), [failed(1)], [gaveUp])).toEqual({ + kind: "stuck", + failed: failed(1), + givenUp: gaveUp, + }); + }); + + it("is retrying, not stuck, once the requester sent a stuck task to another agent", () => { + expect(taskProgress(task(["1"]), [failed(1)], [sending(2)]).kind).toBe("retrying"); + }); + + it("ignores a dispatch older than the newest attempt", () => { + expect(taskProgress(task(["1"]), [failed(1)], [sending(1, { state: "given_up" })]).kind).toBe("stuck"); + expect(taskProgress(task(), [], [sending(1, { state: "escrowing" })])).toEqual({ kind: "waiting" }); + }); }); describe("attemptOutcome", () => { diff --git a/frontend/src/screens/live/run/progress.ts b/frontend/src/screens/live/run/progress.ts index e25d0b3d..af39659f 100644 --- a/frontend/src/screens/live/run/progress.ts +++ b/frontend/src/screens/live/run/progress.ts @@ -1,20 +1,21 @@ -import type { Attempt, MoneyEvent, PlanTask } from "../../../api/types"; +import type { Attempt, MoneyEvent, PendingDispatch, PlanTask } from "../../../api/types"; /** - * What each task of a running plan is doing, derived from its attempts and the requester's agent picks, - * the same rule the control plane applies (plan task O6): a failed or inconclusive attempt is retried with - * the next agent the requester picked; once every pick has been tried, the task waits on the requester. - * The API does not store that last state, so the screen derives it the way the server does. + * What each task of a running plan is doing, derived from its attempts, the attempts dispatched but not yet + * started, and the requester's agent picks, the same rule the control plane applies (plan task O6, + * `outcomes.stuck_indexes`): a failed or inconclusive attempt is retried with the next agent the requester + * picked; once every pick has been tried, or the next attempt could not be started, the task waits on the + * requester. The API does not store that last state, so the screen derives it the way the server does. */ export type TaskProgress = /** Not dispatched yet: waiting on a dependency, or on its escrow. */ | { kind: "waiting" } | { kind: "running"; attempt: Attempt } | { kind: "passed"; attempt: Attempt } - /** The last attempt did not pass and the next pick is being dispatched. */ - | { kind: "retrying"; failed: Attempt; nextAgent: string } - /** Every agent the requester picked has been tried: only the requester can move this task on. */ - | { kind: "stuck"; failed: Attempt }; + /** The last attempt did not pass and the next pick is on its way; `pending` says where, once known. */ + | { kind: "retrying"; failed: Attempt; nextAgent: string; pending?: PendingDispatch } + /** Only the requester can move this task on: every pick was tried, or `givenUp` never started. */ + | { kind: "stuck"; failed?: Attempt; givenUp?: PendingDispatch }; export type AttemptOutcome = "in progress" | "passed" | "failed" | "inconclusive"; @@ -28,21 +29,28 @@ export function attemptOutcome(attempt: Attempt): AttemptOutcome { return attempt.phi_holds === false ? "failed" : "inconclusive"; } -export function taskProgress(task: PlanTask, attempts: Attempt[]): TaskProgress { +export function taskProgress(task: PlanTask, attempts: Attempt[], pending: PendingDispatch[] = []): TaskProgress { const mine = attemptsOf(task.index, attempts); const passed = mine.find((a) => a.phi_holds === true); if (passed) return { kind: "passed", attempt: passed }; const latest = mine.at(-1); + // An attempt dispatched after the newest job: on its way to the gateway, or given up before it started. + const next = pending + .filter((p) => p.index === task.index && p.attempt > (latest?.attempt ?? 0)) + .sort((a, b) => a.attempt - b.attempt) + .at(-1); + if (next?.state === "given_up") return { kind: "stuck", failed: latest, givenUp: next }; if (!latest) return { kind: "waiting" }; if (attemptOutcome(latest) === "in progress") return { kind: "running", attempt: latest }; + if (next) return { kind: "retrying", failed: latest, nextAgent: next.agent_id, pending: next }; // Attempt n ran with agents[n - 1], so the retry's pick is agents[n]. Approval caps the list at the // service's attempt limit, so the list running out is the limit. const nextAgent = task.agents?.[latest.attempt]; return nextAgent === undefined ? { kind: "stuck", failed: latest } : { kind: "retrying", failed: latest, nextAgent }; } -export function stuckTasks(tasks: PlanTask[], attempts: Attempt[]): PlanTask[] { - return tasks.filter((t) => taskProgress(t, attempts).kind === "stuck"); +export function stuckTasks(tasks: PlanTask[], attempts: Attempt[], pending: PendingDispatch[] = []): PlanTask[] { + return tasks.filter((t) => taskProgress(t, attempts, pending).kind === "stuck"); } /** Nothing was ever dispatched: every task still waits. A closed plan like this never funded, or was closed first. */ diff --git a/frontend/src/styles/screens/live-run.css b/frontend/src/styles/screens/live-run.css index 0a96f06f..a124d6cc 100644 --- a/frontend/src/styles/screens/live-run.css +++ b/frontend/src/styles/screens/live-run.css @@ -101,6 +101,15 @@ border-color: color-mix(in srgb, var(--amber) 45%, var(--border)); background: var(--amber-wash); } +.lr-retry-row { + flex-wrap: wrap; +} +/* The retry picker sizes to its agent names, not to the banner's width. */ +.lr-select.lr-retry-select { + width: auto; + min-width: min(280px, 100%); + max-width: 100%; +} .lr-list { list-style: none; } From 15fb7035a171c1d6159fd025045ca7addb59f0dd Mon Sep 17 00:00:00 2001 From: OfficialAbhinavSingh Date: Fri, 6 Nov 2026 12:16:31 +0530 Subject: [PATCH 2/2] control-plane, frontend: retries lock with a full deadline, never twice, and never silently Review of the previous commit found four ways a retry still went wrong: - Its deadline was counted from when it was queued, but it can only lock once the failed attempt's payment is back, up to that attempt's deadline. It locked with almost none left, and the keeper returned it before a worker claimed the job. The deadline now counts from the previous attempt's deadline plus RETURN_MARGIN_S. - A lock that reverted (plan closed, balance short) arrives as a gateway `error` event, which was only logged, so the task said "Retrying" forever. That event now marks its dispatch given up, and the task waits on the requester. - An older attempt whose lock landed after its dispatch looked given up re-dispatched an attempt number the requester's retry already took. The unique-key error rolled back the whole outcome pass, for every plan. Only the newest attempt retries on its own now, and a task that passes cancels its dispatches still on their way. - The gateway only waits on attempt n-1. Behind a given-up attempt, a retry could lock a second payment while an older one still held one. The retry endpoint refuses until the ledger shows that payment released or returned (money.holding_attempt, kept in the ledger module so the graph side still never reads it). The run page takes the stuck list from the plan detail instead of deriving it, so no Retry is offered while the orchestrator is still reacting to a finished job, and hides Retry at max_requester_attempts. Migration 0023 indexes the outbox by a dispatch's task_id, which these lookups join on. --- control-plane/mergit_cp/chain_events.py | 10 +- control-plane/mergit_cp/dispatch.py | 40 +++-- control-plane/mergit_cp/money.py | 23 +++ control-plane/mergit_cp/outcomes.py | 20 ++- control-plane/mergit_cp/routes_plans.py | 8 +- control-plane/mergit_cp/routes_runs.py | 1 + .../migrations/0023_outbox_dispatch_task.sql | 5 + control-plane/tests/test_outcomes.py | 150 +++++++++++++++++- frontend/src/api/types.ts | 4 + .../src/screens/live/run/LiveRunWorkspace.tsx | 9 +- frontend/src/screens/live/run/RunView.tsx | 17 +- .../src/screens/live/run/progress.test.ts | 58 ++++++- frontend/src/screens/live/run/progress.ts | 22 ++- 13 files changed, 336 insertions(+), 31 deletions(-) create mode 100644 control-plane/migrations/0023_outbox_dispatch_task.sql diff --git a/control-plane/mergit_cp/chain_events.py b/control-plane/mergit_cp/chain_events.py index ecacb3db..c8f707bb 100644 --- a/control-plane/mergit_cp/chain_events.py +++ b/control-plane/mergit_cp/chain_events.py @@ -5,7 +5,9 @@ that drive the task graph are acted on here. The money and verdict events (``receipt.committed``, ``verdict.recorded``, ``task.released``, ``task.returned``, ``challenge.*``, ``credits.denied``) are written to the ``money_events`` ledger by ``mergit_cp/money.py`` in the same transaction (plan task O6, -money half); they move money and reputation, never the graph. ``error`` stays only in the raw log. +money half); they move money and reputation, never the graph. An ``error`` about a ``task.dispatch_requested`` +(its lock reverted: the plan was closed, or its balance was short) marks that dispatch given up, since no job +will ever come for it; any other ``error`` stays only in the raw log. """ from __future__ import annotations @@ -84,6 +86,12 @@ async def receive(conn: AsyncConnection[TupleRow], envelope: dict[str, Any], set await on_plan_funded(conn, payload, settings) elif envelope["type"] == "task.escrowed": await on_task_escrowed(conn, payload) + elif envelope["type"] == "error": + await conn.execute( + "UPDATE outbox SET failed_at = coalesce(failed_at, now()), last_error = %s" + " WHERE event_id::text = %s AND type = 'task.dispatch_requested'", + (f"{payload['code']}: {payload['message']}", payload["ref_event_id"]), + ) elif envelope["type"] in money.KINDS: await money.record(conn, envelope) return True diff --git a/control-plane/mergit_cp/dispatch.py b/control-plane/mergit_cp/dispatch.py index 1fae56d4..b5c1cf70 100644 --- a/control-plane/mergit_cp/dispatch.py +++ b/control-plane/mergit_cp/dispatch.py @@ -29,10 +29,35 @@ DISPATCH_DEADLINE_S = 1800 """How long a dispatched task has to be sealed before the plan may give up on it (spec/events-v0.md).""" +RETURN_MARGIN_S = 300 +"""After an attempt's deadline, how long until its payment is surely back in the plan: the gateway +keeper's slack (60s) and tick, then the outbox backoff before the next delivery (up to 60s).""" + RETRY_WINDOW_S = DISPATCH_DEADLINE_S + 600 -"""How long a retry's dispatch keeps being delivered. The gateway refuses to lock attempt n while attempt n-1 -still holds its payment (503), and an attempt that ended without a receipt only gets it back once its deadline -passes and the keeper returns it. A shorter window gives the retry up before then, for good.""" +"""The least a retry's dispatch keeps being delivered for. The gateway refuses to lock attempt n while +attempt n-1 still holds its payment (503), and an attempt that ended without a receipt only gets it back +once its deadline passes and the keeper returns it. A shorter window gives the retry up before then.""" + + +async def _retry_timing( + conn: AsyncConnection[TupleRow], plan: StoredPlan, index: int, attempt: int +) -> tuple[int, int]: + """(deadline_unix, retry window) for attempt ``attempt``. A retry can only lock once attempt n-1's + payment is back, which may be up to that attempt's deadline: counting this attempt's deadline from now + would let it lock with almost none of it left, and the keeper would return it before a worker even + claimed the job.""" + now = int(time.time()) + if attempt == 1: + return now + DISPATCH_DEADLINE_S, 0 + cur = await conn.execute( + "SELECT max((body->'payload'->>'deadline_unix')::bigint) FROM outbox" + " WHERE type = 'task.dispatch_requested' AND body->'payload'->>'task_id' = %s", + (mergit_receipt.task_id(plan.plan_id, index, attempt - 1),), + ) + row = await cur.fetchone() + previous = None if row is None else row[0] + earliest_lock = now if previous is None else max(now, int(previous) + RETURN_MARGIN_S) + return earliest_lock + DISPATCH_DEADLINE_S, max(RETRY_WINDOW_S, earliest_lock - now + 600) async def _task_states(conn: AsyncConnection[TupleRow], plan_uuid: str) -> list[TaskState]: @@ -70,6 +95,7 @@ async def dispatch_task( if task.phi_hash is None: raise ValueError(f"task {index} of plan {plan.plan_id} has no phi_hash; the plan is not approved") task_id = mergit_receipt.task_id(plan.plan_id, index, attempt) + deadline_unix, retry_window_s = await _retry_timing(conn, plan, index, attempt) await conn.execute( 'INSERT INTO task_dispatches (task_id, plan_uuid, "index", attempt, agent_id)' " VALUES (%s, %s, %s, %s, %s)", @@ -87,14 +113,10 @@ async def dispatch_task( "amount_usdc_units": str(task.draft.budget_usdc_units), "category": plan.category, "phi_hash": task.phi_hash, - "deadline_unix": int(time.time()) + DISPATCH_DEADLINE_S, + "deadline_unix": deadline_unix, }, plan_id=plan.plan_id, - retry_window_s=( - settings.event_retry_window_s - if attempt == 1 - else max(settings.event_retry_window_s, RETRY_WINDOW_S) - ), + retry_window_s=max(settings.event_retry_window_s, retry_window_s), ) return task_id diff --git a/control-plane/mergit_cp/money.py b/control-plane/mergit_cp/money.py index 4322f5d4..beb9875b 100644 --- a/control-plane/mergit_cp/money.py +++ b/control-plane/mergit_cp/money.py @@ -246,3 +246,26 @@ def view(e: MoneyEvent) -> dict[str, Any]: "tx": e.tx, "payload": payload(e), } + + +_HOLDING = """ +SELECT max(j.attempt) FROM jobs j +WHERE j.plan_uuid = %(plan)s AND j."index" = %(index)s + AND NOT EXISTS (SELECT 1 FROM jobs n + WHERE n.plan_uuid = %(plan)s AND n."index" = %(index)s + AND n.attempt = (SELECT max(attempt) FROM task_dispatches + WHERE plan_uuid = %(plan)s AND "index" = %(index)s)) + AND NOT EXISTS (SELECT 1 FROM money_events m + WHERE m.task_id = j.task_id AND m.kind IN ('task.released', 'task.returned')) +""" + + +async def holding_attempt(conn: AsyncConnection[TupleRow], plan_uuid: UUID | str, index: int) -> int | None: + """An earlier attempt of task ``index`` whose payment the ledger has not seen released or returned, when + the newest dispatch of that task never locked one (it has no job). The gateway only waits on attempt n-1 + before locking attempt n, so a retry sent behind a given-up attempt would lock a second payment out of + the plan's shared balance while this one is still held. None when a retry is safe: the newest attempt + locked (the gateway waits on it), or nothing earlier holds money. It only reads the graph's tables.""" + cur = await conn.execute(_HOLDING, {"plan": str(plan_uuid), "index": index}) + row = await cur.fetchone() + return None if row is None or row[0] is None else int(row[0]) diff --git a/control-plane/mergit_cp/outcomes.py b/control-plane/mergit_cp/outcomes.py index 80d4a2e4..e3d55fa3 100644 --- a/control-plane/mergit_cp/outcomes.py +++ b/control-plane/mergit_cp/outcomes.py @@ -132,7 +132,19 @@ async def process_one(conn: AsyncConnection[TupleRow], job: FinishedJob, setting await _mark_processed(conn, job.task_id) return - if job.holds is not True: + if job.holds is True: + # Nothing still on its way to the gateway for this index may lock more money once it passed + # (a requester retry sent while this attempt's own dispatch looked given up, say). + await conn.execute( + "UPDATE outbox SET failed_at = now(), last_error = 'the task already passed'" + " WHERE plan_id = %s AND type = 'task.dispatch_requested'" + " AND (body->'payload'->>'index')::int = %s" + " AND delivered_at IS NULL AND failed_at IS NULL", + (plan.plan_id, job.index), + ) + elif job.attempt == await next_attempt(conn, plan, job.index) - 1: + # Only the newest attempt retries on its own: an older one finishing late (its dispatch looked + # given up and the requester already sent a newer one) must not reuse a taken attempt number. next_agent = _next_agent(task, job.attempt, settings.max_attempts) if next_agent is not None: await dispatch.dispatch_task(conn, plan, job.index, job.attempt + 1, next_agent, settings) @@ -205,9 +217,9 @@ async def retry( ) -> str: """The requester sends a stuck index to one more agent. The agent takes slot ``attempt`` of ``tasks.agents`` (any later pick a given-up dispatch never reached is dropped), so attempt ``n`` keeps - naming ``agents[n - 1]`` as everywhere else. It is paid from the index's own escrow, which comes back to - the plan when the failed attempt's money is returned (the gateway holds the dispatch until then). The - caller checks the index is stuck and the agent exists.""" + naming ``agents[n - 1]`` as everywhere else. It is paid from the plan's escrow once the failed attempt's + payment is back in it (the gateway holds the dispatch until then). The caller checks the index is stuck, + that no earlier attempt still holds money (``money.holding_attempt``), and that the agent exists.""" attempt = await next_attempt(conn, plan, index) await conn.execute( "UPDATE tasks SET agents = coalesce(agents[1:%s], ARRAY[]::text[]) || ARRAY[%s::text]" diff --git a/control-plane/mergit_cp/routes_plans.py b/control-plane/mergit_cp/routes_plans.py index 88b37132..4abe9e7f 100644 --- a/control-plane/mergit_cp/routes_plans.py +++ b/control-plane/mergit_cp/routes_plans.py @@ -17,7 +17,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response from pydantic import BaseModel, ConfigDict, Field, StringConstraints -from . import accounts, outcomes, plans, redeem, repo_check, user_settings +from . import accounts, money, outcomes, plans, redeem, repo_check, user_settings from .agents import Agent from .auth import PoolDep, Principal, PrincipalDep, SettingsDep from .events import enqueue @@ -371,6 +371,12 @@ async def retry_task( raise _problems(422, [f"agent {body.agent_id} is not deployed"]) if index not in await outcomes.stuck_indexes(conn, plan, settings): raise HTTPException(status_code=409, detail=f"task {index} is not waiting on you") + holding = await money.holding_attempt(conn, plan.plan_uuid, index) + if holding is not None: + raise HTTPException( + status_code=409, + detail=f"attempt {holding} of task {index} still holds its payment; retry once it is back", + ) attempt = await outcomes.next_attempt(conn, plan, index) if attempt > settings.max_requester_attempts: raise HTTPException( diff --git a/control-plane/mergit_cp/routes_runs.py b/control-plane/mergit_cp/routes_runs.py index df5d7d91..532bd832 100644 --- a/control-plane/mergit_cp/routes_runs.py +++ b/control-plane/mergit_cp/routes_runs.py @@ -78,6 +78,7 @@ async def plan_detail( view["money_events"] = [money.view(m) for m in ledger] view["pending_dispatches"] = [runs.pending_view(p) for p in pending] view["stuck_indexes"] = stuck + view["max_requester_attempts"] = settings.max_requester_attempts return view diff --git a/control-plane/migrations/0023_outbox_dispatch_task.sql b/control-plane/migrations/0023_outbox_dispatch_task.sql new file mode 100644 index 00000000..f0f49aca --- /dev/null +++ b/control-plane/migrations/0023_outbox_dispatch_task.sql @@ -0,0 +1,5 @@ +-- A dispatch's outbox row, by the task_id its payload names. `outcomes.stuck_indexes` and `runs.pending_dispatches` +-- join `task_dispatches` to the outbox on it for every plan detail and every running plan in the plan list, and the +-- run page reloads the detail while a retry waits; outbox rows are never deleted, so without this each lookup scans +-- the whole outbox. +CREATE INDEX outbox_dispatch_task ON outbox ((body->'payload'->>'task_id')) WHERE type = 'task.dispatch_requested'; diff --git a/control-plane/tests/test_outcomes.py b/control-plane/tests/test_outcomes.py index b215c825..1af09e3e 100644 --- a/control-plane/tests/test_outcomes.py +++ b/control-plane/tests/test_outcomes.py @@ -17,7 +17,7 @@ from psycopg.rows import TupleRow from pydantic import SecretStr -from mergit_cp import chain_events, outcomes, plans +from mergit_cp import chain_events, money, outcomes, plans from mergit_cp.config import Settings from mergit_cp.db import Pool, migrate, open_pool from mergit_cp.planner import TaskDraft @@ -522,3 +522,151 @@ async def body(pool: Pool) -> None: assert loaded is not None and loaded.tasks[0].agents == ["1", "2", "1"] run(database_url, body) + + +def _event(event_type: str, payload: dict[str, Any]) -> dict[str, Any]: + return { + "event_id": str(uuid.uuid4()), + "type": event_type, + "at": "2026-11-06T00:00:00.000Z", + "payload": payload, + } + + +async def _deadline(conn: psycopg.AsyncConnection[TupleRow], attempt: int) -> int: + cur = await conn.execute( + "SELECT (body->'payload'->>'deadline_unix')::bigint FROM outbox" + " WHERE type = 'task.dispatch_requested' AND (body->'payload'->>'attempt')::int = %s", + (attempt,), + ) + row = await cur.fetchone() + assert row is not None + return int(row[0]) + + +def test_a_retrys_deadline_counts_from_when_it_can_lock(database_url: str) -> None: + """Attempt 2 locks no earlier than attempt 1's deadline (its payment comes back then), so its own deadline + is a full DISPATCH_DEADLINE_S after that, not after the moment it was queued.""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + async with pool.connection() as conn: + first, second = await _deadline(conn, 1), await _deadline(conn, 2) + window = await _window_s(conn, 2) + d = outcomes.dispatch + assert second >= first + d.RETURN_MARGIN_S + d.DISPATCH_DEADLINE_S + # And it keeps being delivered past the moment it can first lock. + assert window >= d.DISPATCH_DEADLINE_S + d.RETURN_MARGIN_S + + run(database_url, body) + + +def test_a_reverted_lock_gives_the_dispatch_up_so_the_requester_can_act(database_url: str) -> None: + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + async with pool.connection() as conn, conn.transaction(): + row = await ( + await conn.execute( + "UPDATE outbox SET delivered_at = now()" + " WHERE (body->'payload'->>'attempt')::int = 2 RETURNING event_id::text" + ) + ).fetchone() + assert row is not None + assert await outcomes.stuck_indexes(conn, plan, SETTINGS) == [] # delivered: waiting on its lock + error = {"ref_event_id": row[0], "code": "TX_REVERTED", "message": "InsufficientPlanBalance"} + await chain_events.receive(conn, _event("error", error), SETTINGS) + assert await outcomes.stuck_indexes(conn, plan, SETTINGS) == [0] + + run(database_url, body) + + +def test_an_older_attempt_finishing_late_dispatches_nothing(database_url: str) -> None: + """Attempt 1's dispatch looked given up, the requester sent attempt 2, then attempt 1's lock landed after + all and its job finished: it must not try to dispatch attempt 2 a second time (a unique-key error there + would roll back the whole outcome pass, for every plan).""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + first = await _dispatch_and_escrow(conn, plan, 0, 1, "1") + await outcomes.dispatch.dispatch_task(conn, plan, 0, 2, "2", SETTINGS) + await _fail(conn, first) + assert await _process(pool) == 1 + async with pool.connection() as conn: + n = await (await conn.execute("SELECT count(*) FROM task_dispatches")).fetchone() + assert n is not None and n[0] == 2 + + run(database_url, body) + + +def test_a_pass_cancels_the_tasks_dispatches_still_on_their_way(database_url: str) -> None: + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + first = await _dispatch_and_escrow(conn, plan, 0, 1, "1") + await conn.execute("UPDATE outbox SET delivered_at = now()") + await outcomes.dispatch.dispatch_task(conn, plan, 0, 2, "2", SETTINGS) + await conn.execute( + "UPDATE jobs SET status = 'done', phi_holds = true, finished_at = now() WHERE task_id = %s", + (first,), + ) + await _process(pool) + async with pool.connection() as conn: + row = await ( + await conn.execute( + "SELECT failed_at IS NOT NULL, last_error FROM outbox" + " WHERE (body->'payload'->>'attempt')::int = 2" + ) + ).fetchone() + assert row == (True, "the task already passed") + + run(database_url, body) + + +def test_no_retry_behind_a_given_up_attempt_while_an_older_one_holds_money(database_url: str) -> None: + """The gateway only checks attempt n-1 before locking n: with attempt 2 given up and attempt 1 still + locked, attempt 3 would take a second payment out of the plan's shared balance.""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1", "2"]}) + first = await _dispatch_and_escrow(conn, plan, 0, 1, "1") + await _fail(conn, first) + await _process(pool) + async with pool.connection() as conn, conn.transaction(): + await conn.execute( + "UPDATE outbox SET failed_at = now() WHERE (body->'payload'->>'attempt')::int = 2" + ) + assert await outcomes.stuck_indexes(conn, plan, SETTINGS) == [0] + assert await money.holding_attempt(conn, plan.plan_uuid, 0) == 1 + returned = { + "task_id": first, + "amount_usdc_units": "500000", + "reason": "deadline", + "tx": "0x" + "d" * 64, + } + await chain_events.receive(conn, _event("task.returned", returned), SETTINGS) + assert await money.holding_attempt(conn, plan.plan_uuid, 0) is None + + run(database_url, body) + + +def test_a_retry_behind_a_locked_attempt_is_safe(database_url: str) -> None: + """Attempt 1 ran and failed: its payment may still be locked, but the gateway itself holds attempt 2 until + it is back, so the retry is allowed.""" + + async def body(pool: Pool) -> None: + async with pool.connection() as conn, conn.transaction(): + plan = await _plan(conn, [task(0)], {0: ["1"]}) + await _fail(conn, await _dispatch_and_escrow(conn, plan, 0, 1, "1")) + await _process(pool) + async with pool.connection() as conn: + assert await money.holding_attempt(conn, plan.plan_uuid, 0) is None + + run(database_url, body) diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 5cd3acb5..e9ee999a 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -78,6 +78,10 @@ export type PlanDetail = Plan & { attempts: Attempt[]; money_events: MoneyEvent[]; pending_dispatches?: PendingDispatch[]; + /** The tasks only the requester can move on, as the control plane decides it (`outcomes.stuck_indexes`). */ + stuck_indexes?: number[]; + /** Attempts one task may reach in all, counting the requester's own retries. */ + max_requester_attempts?: number; }; export type PlanSummary = Pick< diff --git a/frontend/src/screens/live/run/LiveRunWorkspace.tsx b/frontend/src/screens/live/run/LiveRunWorkspace.tsx index 0e8012e7..8a2c6321 100644 --- a/frontend/src/screens/live/run/LiveRunWorkspace.tsx +++ b/frontend/src/screens/live/run/LiveRunWorkspace.tsx @@ -7,7 +7,7 @@ import type { AgentInfo, PlanDetail, StepEvent } from "../../../api/types"; import { MergitLoader } from "../../../components/MergitLoader"; import { PageHeader } from "../../../components/PageHeader"; import { PlanBadge } from "./badges"; -import { stuckTasks, taskProgress } from "./progress"; +import { taskProgress, waitingOnRequester } from "./progress"; import { DraftPlanEditor } from "./DraftPlanEditor"; import { FundingStep } from "./FundingStep"; import { RunView } from "./RunView"; @@ -64,12 +64,7 @@ export function LiveRunWorkspace() { {detail.goal}} crumb={{ label: "Runs", to: paths.runs, id: detail.plan_uuid.slice(0, 8) }} - badge={ - - } + badge={} subtitle={ <> {detail.repo_url} at {detail.ref} diff --git a/frontend/src/screens/live/run/RunView.tsx b/frontend/src/screens/live/run/RunView.tsx index 426976fa..0d4a9f71 100644 --- a/frontend/src/screens/live/run/RunView.tsx +++ b/frontend/src/screens/live/run/RunView.tsx @@ -8,7 +8,7 @@ import { ProgressBadge } from "./badges"; import { ClosePlan } from "./ClosePlan"; import { PullRequestStep } from "./PullRequestStep"; import { RetryTask } from "./RetryTask"; -import { attemptsOf, closedBeforeStart, stuckTasks, taskProgress } from "./progress"; +import { attemptsOf, closedBeforeStart, lastAttempt, taskProgress, waitingOnRequester } from "./progress"; import { canStartAgain } from "./startAgain"; import { StepLog } from "./StepLog"; import { TaskFlow } from "./TaskFlow"; @@ -164,7 +164,8 @@ function RunBanner({
); } - const stuck = stuckTasks(detail.tasks, detail.attempts, detail.pending_dispatches ?? []); + const stuck = waitingOnRequester(detail); + const limit = detail.max_requester_attempts; if (stuck.length === 0) return null; const names = stuck.map((t) => `task ${String(t.index + 1)}`).join(", "); return ( @@ -176,9 +177,15 @@ function RunBanner({ Nothing more runs on its own. Retry a task with another agent (it is paid from that task's own budget), or close the plan to stop it and get its unspent funds back. - {stuck.map((t) => ( - - ))} + {stuck.map((t) => + limit !== undefined && lastAttempt(t.index, detail) >= limit ? ( + + Task {t.index + 1} has had its last attempt ({limit}). + + ) : ( + + ), + )} ); diff --git a/frontend/src/screens/live/run/progress.test.ts b/frontend/src/screens/live/run/progress.test.ts index 7e186de0..402afefd 100644 --- a/frontend/src/screens/live/run/progress.test.ts +++ b/frontend/src/screens/live/run/progress.test.ts @@ -1,6 +1,15 @@ import { describe, expect, it } from "vitest"; -import type { Attempt, MoneyEvent, PendingDispatch, PlanTask } from "../../../api/types"; -import { attemptOutcome, closedBeforeStart, creditsDenial, stuckTasks, taskProgress, verdictFor } from "./progress"; +import type { Attempt, MoneyEvent, PendingDispatch, PlanDetail, PlanTask } from "../../../api/types"; +import { + attemptOutcome, + closedBeforeStart, + creditsDenial, + lastAttempt, + stuckTasks, + taskProgress, + verdictFor, + waitingOnRequester, +} from "./progress"; const task = (agents: string[] | null = ["1", "2"]): PlanTask => ({ index: 0, @@ -156,3 +165,48 @@ describe("verdictFor and creditsDenial", () => { expect(creditsDenial([])).toBeNull(); }); }); + +describe("waitingOnRequester", () => { + const detail = (over: Partial): PlanDetail => ({ + plan_uuid: "p", + plan_id: "0x1", + requester: "0x2", + state: "running", + goal: "", + repo_url: "", + ref: "main", + category: "test-repair", + budget_usdc_units: "500000", + funding: "credits", + total_budget_usdc_units: "500000", + created_at: "", + approved_at: null, + tasks: [task(["1"])], + attempts: [failed(1)], + money_events: [], + ...over, + }); + + it("trusts the control plane's list when it sent one", () => { + // The job finished but the orchestrator has not reacted yet: the screen alone would call it stuck. + expect(waitingOnRequester(detail({ stuck_indexes: [] }))).toEqual([]); + expect(waitingOnRequester(detail({ stuck_indexes: [0] }))).toEqual([task(["1"])]); + }); + + it("derives it the server's way when the list is missing", () => { + expect(waitingOnRequester(detail({}))).toEqual([task(["1"])]); + }); + + it("counts attempts on their way as well as the ones that ran", () => { + const pending: PendingDispatch = { + index: 0, + attempt: 3, + agent_id: "2", + state: "sending", + last_error: null, + give_up_at: null, + }; + expect(lastAttempt(0, detail({ pending_dispatches: [pending] }))).toBe(3); + expect(lastAttempt(1, detail({}))).toBe(0); + }); +}); diff --git a/frontend/src/screens/live/run/progress.ts b/frontend/src/screens/live/run/progress.ts index af39659f..b100a6ab 100644 --- a/frontend/src/screens/live/run/progress.ts +++ b/frontend/src/screens/live/run/progress.ts @@ -1,4 +1,4 @@ -import type { Attempt, MoneyEvent, PendingDispatch, PlanTask } from "../../../api/types"; +import type { Attempt, MoneyEvent, PendingDispatch, PlanDetail, PlanTask } from "../../../api/types"; /** * What each task of a running plan is doing, derived from its attempts, the attempts dispatched but not yet @@ -53,6 +53,26 @@ export function stuckTasks(tasks: PlanTask[], attempts: Attempt[], pending: Pend return tasks.filter((t) => taskProgress(t, attempts, pending).kind === "stuck"); } +/** + * The tasks waiting on the requester. The control plane's own list when it sent one: between a job finishing + * and the orchestrator reacting to it, the screen's derivation alone would call a task stuck that is about to + * be retried, and offer a Retry the server refuses. + */ +export function waitingOnRequester(detail: PlanDetail): PlanTask[] { + const fromServer = detail.stuck_indexes; + return fromServer + ? detail.tasks.filter((t) => fromServer.includes(t.index)) + : stuckTasks(detail.tasks, detail.attempts, detail.pending_dispatches ?? []); +} + +/** The highest attempt number this task has had or has on its way. */ +export function lastAttempt(index: number, detail: PlanDetail): number { + const numbers = [...detail.attempts, ...(detail.pending_dispatches ?? [])] + .filter((a) => a.index === index) + .map((a) => a.attempt); + return Math.max(0, ...numbers); +} + /** Nothing was ever dispatched: every task still waits. A closed plan like this never funded, or was closed first. */ export function closedBeforeStart(tasks: PlanTask[], attempts: Attempt[]): boolean { return tasks.every((t) => taskProgress(t, attempts).kind === "waiting");