The harness, in code · Build it

No-progress detector

Implement the check that names the turn an agent loop starts repeating itself, catching two-call cycles and reordered arguments but not a retry that worked.

codemediumExercise · 3 min · 70 of 77

On this page

A stuck loop is cheap per turn and expensive in total, and neither cap notices it quickly. The no-progress check is the exit that does, provided it can tell a loop going round from a loop recovering.

The situation

A team's support agent stops when the same tool is called with the same arguments twice in a row, the strict check from agent loops and stop conditions. Over 2,000 runs it missed three kinds of stuck loop and stopped one kind of healthy one. 17 runs alternated between get_order(123) and get_shipment(123) until the turn cap of 25. 4 cycled through three calls. 6 repeated one call with its argument keys reordered, {"id": 123, "full": true} then {"full": true, "id": 123}, which a raw-string comparison counts as two different calls. And 9 healthy runs were stopped because they retried a call once after a timeout and got an answer the second time.

Each stuck run spent about 20 turns it did not need. The harness wants one function that names the turn at which the loop is first seen repeating, with a window of recent turns as its only memory.

The question

Implement the detector so that the strict repeat, the two-call cycle and the reordered keys all fire on the turn the repetition completes, and the retry after a timeout does not.

def solve(calls, window):
    """calls: one (name, args, result) per turn, oldest first; turn i is
    calls[i - 1]. window: how many recent turns the detector keeps, so it can
    see a cycle of up to window // 2 turns.
    Returns the first turn t at which the last 2p turns are one block of p
    turns played twice, for some p with 1 <= p <= window // 2, or None if that
    never happens. The same call with a different result is progress."""

S = ("search_tickets", {"q": "refund delay"}, "3 tickets: 811, 812, 840")
A = ("get_order", {"id": 123}, '{"status": "shipped"}')
B = ("get_shipment", {"id": 123}, '{"eta": "Friday"}')

solve([S, S, S], 2)                                                       -> 2
solve([A, B, A, B, A, B], 2)                                              -> None
solve([A, B, A, B, A, B], 4)                                              -> 4
solve([("get_order", {"id": 123}, "error: timed out after 5 s"), A], 2)   -> None

A strong answer · Rubric · Where answers go wrong · Solution

Write your answer first, then open

A strong answer

Key each turn canonically: the tool name, the arguments serialised with sorted keys and fixed separators, and the result. Sorted keys make the reordered call the same call. The result makes a retry after a timeout a different turn, since it went from an error to an answer, and makes a status poll that changed into progress; one that returns pending again is not.

Keep the last window keys in a bounded deque. After appending turn t, compare the last p keys with the p before them for each period from 1 to half the keys held, shortest first, and return t on the first match. Period 1 is the strict check; period 2 catches the alternation on turn 4; a window of 6 sees a three-call cycle on turn 6. Return the turn, not a boolean, so the trace records no_progress at turn 4 and the loop stops before sending turn 5. With a window of 6 the work is at most six key comparisons a turn.

Choose the window from the turn cap: a cycle longer than half the cap cannot play twice before the cap fires, and with a cap of 25 a window of 6 covers every cycle the trace showed.

The detector never compares names alone, since a search with new terms is progress, and never fires on two equal calls with other work between them. The evidence that would change the key: tools whose results carry a timestamp or request id, which make every result unique and the detector blind; strip those fields before keying. It is the third exit in a harness in 200 lines.

Rubric

keys each turn on name, canonical arguments and result, so reordered keys match and a retry with a new result does not
checks every period from 1 to window // 2 and fires on the turn the second copy of the block completes
returns the turn number, so the trace can say where the loop stopped
holds only the last window keys, so memory does not grow with the run
returns None for a run that never repeats, including an empty one

Where answers go wrong

"Compare the last two calls." That is the check that shipped. The second example returns None for it, which is the 17 alternating runs reaching the turn cap.

"Hash the name and arguments and leave the result out." The 9 healthy runs are stopped again, and a status poll that moves from pending to done reads as a loop going nowhere.

"Count each call across the whole run and stop at the third." A run that re-reads config.yaml after each of three edits is making progress, since the file changed each time, and the counter stops it; the alternation, meanwhile, runs to turn 5 before either call is counted three times.

Solution

Python 3 · standard library33 lines · 13 of 13 tests passing

"""No-progress detector — the turn at which an agent loop is first seen repeating itself."""
import json
from collections import deque


def key(call):
    """Canonical form of one turn: tool, arguments and result. Key order and spacing never matter."""
    name, args, result = call
    return json.dumps([name, args, result], sort_keys=True, separators=(",", ":"))


def solve(calls, window):
    """calls: one (name, args, result) per turn, oldest first; turn i is
    calls[i - 1]. window: how many recent turns the detector keeps, so it can
    see a cycle of up to window // 2 turns.
    Returns the first turn t at which the last 2p turns are one block of p
    turns played twice, for some p with 1 <= p <= window // 2, or None if that
    never happens. The same call with a different result is progress."""
    recent = deque(maxlen=window)           # memory is bounded by the window, not by the run
    for t, call in enumerate(calls, 1):
        recent.append(key(call))
        r, m = list(recent), len(recent)
        # Every cycle length the window can hold, shortest first: p = 1 is "same call twice in a row".
        for p in range(1, m // 2 + 1):
            if r[m - 2 * p : m - p] == r[m - p :]:
                return t                    # the turn the repetition completes; stop before sending t + 1
    return None


S = ("search_tickets", {"q": "refund delay"}, "3 tickets: 811, 812, 840")
A = ("get_order", {"id": 123}, '{"status": "shipped"}')
B = ("get_shipment", {"id": 123}, '{"eta": "Friday"}')
C = ("get_invoice", {"id": 123}, '{"total": 40}')

The cases that ran

TESTS = [
    # the examples on the page
    (([S, S, S], 2), 2),
    (([A, B, A, B, A, B], 2), None),
    (([A, B, A, B, A, B], 4), 4),
    (([("get_order", {"id": 123}, "error: timed out after 5 s"), A], 2), None),
    # a poll whose result changed is progress; one whose result did not is not
    (([("get_deploy_status", {"id": "d-91"}, "pending"), ("get_deploy_status", {"id": "d-91"}, "done")], 2), None),
    (([("get_deploy_status", {"id": "d-91"}, "pending")] * 2, 6), 2),
    # argument key order is not a different call
    (([("f", {"a": 1, "b": 2}, "ok"), ("f", {"b": 2, "a": 1}, "ok")], 2), 2),
    # a three-turn cycle needs a window of six
    (([A, B, C, A, B, C], 4), None),
    (([A, B, C, A, B, C], 6), 6),
    # repetition after real progress fires on the turn it completes
    (([S, A, B, A, B], 4), 5),
    # new arguments are progress
    (([("search_tickets", {"q": "refund"}, "0 tickets"), S], 2), None),
    (([], 4), None),
    (([S, S], 1), None),
]