Escalate on a budget, not an error: when an AI agent should hand off to a human

TaskPool AI — Wed, 14 Oct 2026 14:38:42 +0000

A retry is not a fresh draw, and waiting for an exception means waiting past the point where anything was recoverable. Four counters you can read before the failure, a handoff packet, and Python that runs.

Mercor's best agent on APEX-Agents passes 24.0% of its tasks on one attempt. Give it eight attempts and it reaches about 40%. A flat 24% chance per attempt would put eight tries near 89%, so most of those failures belong to the task and not to the dice. That inference is mine. Nothing I read in Mercor's write-up says it, and you are free to argue with it.

The usual answer to a failed run is another run, often resumed from the same conversation. Sinha and colleagues found that models make more mistakes when their own earlier mistakes sit in the context, and that scaling the model does not remove the effect, although thinking models mitigate it. Their tasks were synthetic, with the plan handed to the model, so carrying the result over to a production retry is my extension and not theirs.

That leaves a move most harnesses never budget for: stopping on purpose and giving the work to someone else. The rest of this post covers the trigger logic and the packet, and then what it costs. If you came for code, skip to the section called "Counters that fire before the exception".

Read the stop reason first

I'm putting the objection ahead of the fix. A lot of what looks like an agent giving up is a harness limit doing exactly what it was configured to do, and handing that to a person means paying a stranger to find your bug. I wrote up six causes of agents stopping mid-task on the Taskpool blog, and five of them live in the harness. Run that triage before you wire up anything below.

Counters that fire before the exception

The class below watches four things per tool call: an identical call repeating inside a short window, steps since the last verified checkpoint, total steps, and wall-clock time. It returns a reason string the moment one trips and None otherwise.

from __future__ import annotations

import hashlib
import json
import time
from collections import deque
from dataclasses import dataclass
from typing import Callable, Optional


@dataclass
class Budget:
    max_steps: int           # set from your own successful runs, not a default
    max_seconds: float
    max_repeats: int = 3     # identical tool call (name + args) this many times...
    window: int = 8          # ...inside the last N calls
    stall_steps: int = 12    # steps allowed without a verified checkpoint


class Tripwire:
    def __init__(self, budget: Budget, clock: Callable[[], float] = time.monotonic):
        self.b = budget
        self.clock = clock
        self.t0 = clock()
        self.steps = 0
        self.since_checkpoint = 0
        self.recent: deque[str] = deque(maxlen=budget.window)

    def observe(self, tool: str, args: dict, checkpoint_advanced: bool = False) -> Optional[str]:
        """Call once per tool call. Returns a reason string when it is time to hand off."""
        self.steps += 1
        self.since_checkpoint = 0 if checkpoint_advanced else self.since_checkpoint + 1

        blob = json.dumps([tool, args], sort_keys=True, default=str).encode()
        fp = hashlib.sha256(blob).hexdigest()[:12]
        self.recent.append(fp)

        if self.recent.count(fp) >= self.b.max_repeats:
            return "repeated_action"
        if self.since_checkpoint >= self.b.stall_steps:
            return "checkpoint_stall"
        if self.steps > self.b.max_steps:
            return "step_budget"
        if self.clock() - self.t0 > self.b.max_seconds:
            return "time_budget"
        return None


@dataclass
class Handoff:
    goal: str                # one sentence, readable by someone with no context
    reason: str              # which tripwire fired
    last_verified: str       # last checkpoint that passed its check
    tried: list[str]         # short descriptions, not the transcript
    ask: str                 # the single thing the human has to do
    accept_if: str           # how you will judge what comes back
    deadline_hours: int
    spend_cap_pence: int     # integer minor units, never floats

The thresholds are defaults I picked so the tests had something to assert. They are not measured. max_steps is the one that matters, and I'd start from the longest successful run you have on file, add slack, and then look at which counter actually fires.

The checkpoint counter is the best signal and the most expensive to earn. A checkpoint here means a state something outside the model confirmed, such as a passing test or a row that exists in the database. If you cannot say what a checkpoint is for your task, stall detection is not available to you yet, and defining it will take longer than writing the class.

Arguments are serialised with sorted keys before hashing, so {"a": 1, "b": 2} and {"b": 2, "a": 1} count as the same call. Without that, an agent can loop forever while shuffling its arguments.

What the human is handed

The transcript stays out. It carries the model's own errors, which is what the Sinha result says to keep away from the next attempt, and it costs a person real time to read. The Handoff dataclass above is the replacement: a goal a stranger can follow, which counter fired, the last checkpoint that passed, short descriptions of what was tried, one ask, and a deadline.

Two fields do more than they look like. accept_if forces you to write the check before the work leaves, which is the point where it is cheapest to be strict. spend_cap_pence is an integer because money held as a float eventually produces a bug report.

Wiring it to LangGraph takes one node. Its interrupt docs require a checkpointer and a thread_id, and resuming goes through Command(resume=...).

from dataclasses import asdict
from langgraph.types import interrupt

def escalate(state):
    packet = Handoff(
        goal="Confirm the shop sign matches the listing",
        reason=state["reason"],
        last_verified="listing page parsed",
        tried=["fetched listing twice", "street view unavailable"],
        ask="Photograph the shopfront and sign",
        accept_if="Sign text legible in at least one photo",
        deadline_hours=48,
        spend_cap_pence=1500,
    )
    # Nothing with side effects above this line.
    answer = interrupt(asdict(packet))
    return {"human_result": answer}

The comment is not decoration. The docs say the node restarts from its first line when you resume, so anything above interrupt() runs twice, and they warn against wrapping the call in a bare try/except because the pause is implemented as an exception. I ran this round trip against langgraph 1.2.15 with the in-memory checkpointer and it passes, along with five tests for the counters. In production the checkpointer has to outlive your process, because a person who answers in two days will not be answering a process that is still running.

Not every handoff goes to a stranger

The run is stuck because	Hand it to
The next step needs a decision only your team has authority to make	The on-call engineer or the account owner
The output exists but needs judgement you already employ	An in-house reviewer
The next step needs a person in a particular place, or an independent check of the agent's own claim	An outside verified human

Only the last row is a job for a marketplace. Sending the first two to a stranger gets you an answer from someone with no authority and no context.

If the destination is a job board

I co-founded Taskpool International, which runs a marketplace where AI agents post tasks for verified humans, so weight this section accordingly. The Python above is vendor-neutral and I have not wired it to any marketplace in this post. What follows comes from reading Taskpool's agent docs, not from a run, and each point applies to any board you might use.

Listings are public. In Taskpool's case the task description appears on the public job board, so the ask field must never carry credentials or customer data. Plan for a stranger to read it.

Handoffs have a floor price. The docs set a minimum payout of £5.00 per worker and a 10% platform fee on top, so the cheapest single slot costs £5.50. That number changes which runs are worth escalating at all.

Silence can pay. The docs say a submission still in review seven days after the worker's deadline is accepted automatically and paid. Run your accept_if check from a webhook, not from someone remembering.

A loop that escalates forever is a retry loop that sends you invoices. Cap spend per key as well as per task; Taskpool's docs describe a monthly budget per API key that refuses funding before any money moves.

What to change on Monday

Log the stop reason on every model call before you theorise. It costs you some log volume.
Pick one verified checkpoint per task type. If the task has none, accept that stall detection is off and say so in the runbook.
Set max_steps from your own successful runs and log which counter fires. You will pay for handoffs on runs that would have finished, and nobody has published how often.
Write accept_if before the work leaves, in a form code can check.
Strip secrets from the packet and assume a stranger can read it.
Cap spend per task and per key.

If you want the longer argument for why task length rather than difficulty sets the wall, I made it with the benchmark numbers in Duration, not difficulty, is what breaks agents.

The number I don't have

I know of no published distribution of why production agent runs get handed off, or of how often a budget like this fires on a run that would have succeeded. That false-positive rate decides whether any of this is worth the money, and I have none. If you have traces, I would rather have your numbers than your agreement.

FAQ

When should an AI agent escalate to a human? When a counter trips, or when the next step needs a person on site or an independent check that the agent cannot supply. Do it before the failure, while the state is still recoverable.

Is retrying a failed agent run a good recovery strategy? Not by default. Models can condition on their own earlier errors, an effect thinking models mitigate and scale alone does not remove, so a fresh run seeded with a clean summary is the safer retry.

How does LangGraph pause an agent for human input? Call interrupt() inside a node, with a checkpointer and a thread_id configured. Resume the same thread with Command(resume=...).

I used Claude to draft this post. Every figure links to its source, checked on 14 October 2026, and the Python was run: six tests pass.

DEV Community: TaskPool AI