control-plane, frontend: retries that actually run; retry a stuck task with another agent

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

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)."""

control-plane/mergit_cp/dispatch.py

@@ -29,6 +29,36 @@
 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
+"""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]:
     cur = await conn.execute(
@@ -65,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)",
@@ -82,10 +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,
+        retry_window_s=max(settings.event_retry_window_s, retry_window_s),
     )
     return task_id
 

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

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])

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 (
@@ -116,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)
@@ -150,23 +178,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 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]"
+        ' 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 +238,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",

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
@@ -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,45 @@ 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")
+        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(
+                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}

control-plane/mergit_cp/routes_runs.py

@@ -64,14 +64,21 @@ 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
+    view["max_requester_attempts"] = settings.max_requester_attempts
     return view