A watchdog for LLM streams that go quiet instead of failing

Shipped

This release came out of a review of three weeks of scheduled-job logs on my fitness agent, which generates a morning briefing on a timer. The job was failing roughly half its runs, and the only signal was the absence of a notification. The generation step opened a stream to the model, the stream stopped delivering messages, and nothing ever raised. One attempt, no timeout, no retry.

The fix is three small pieces that fit any streaming model call: a watchdog on the gap between messages, a bounded retry around the whole attempt, and an error message that names the real failure. That is what this guide builds.

Why a total timeout is the wrong tool here

The obvious instinct is to wrap the call in a deadline. Give it five minutes, fail if it runs long.

That works badly for generation. A long response legitimately takes minutes, and the length varies with the input, so any total budget you pick is either so tight it kills good runs or so loose that a dead stream still burns the whole window before you notice. My successful runs completed in 47 to 110 seconds. A dead one sat there for four minutes and produced nothing.

The useful signal is not total elapsed time. It is the gap between consecutive messages. A healthy stream is chatty: the Anthropic streaming API emits message_start, then a run of content_block_delta events, and may emit ping events at any point in the stream. Silence for two minutes in the middle of that is not the model thinking. It is a stream that has stopped.

So set your bound on the quiet period, and pick the number from your own observed latency. Mine is 120 seconds against runs that normally finish in under two minutes, which makes it unambiguous.

Step 1: wrap the stream in an idle watchdog

The whole trick is an async generator that sits between you and the stream. It pulls one message at a time under asyncio.wait_for, so the clock resets on every message rather than running against the whole response.

# watchdog.py
"""An idle watchdog for any async stream."""
from __future__ import annotations

import asyncio


class StreamIdleTimeout(RuntimeError):
    """The stream went silent past the idle timeout, so the run is dead."""


async def iter_with_idle_timeout(source, timeout_s: float):
    """Yield from `source`, raising StreamIdleTimeout when the gap between
    two consecutive messages exceeds `timeout_s`. A timeout of 0 or less
    disables the watchdog and passes messages straight through."""
    it = source.__aiter__()
    try:
        while True:
            try:
                if timeout_s > 0:
                    msg = await asyncio.wait_for(it.__anext__(), timeout=timeout_s)
                else:
                    msg = await it.__anext__()
            except StopAsyncIteration:
                return
            except TimeoutError:
                raise StreamIdleTimeout(
                    f"no stream message for {timeout_s:.0f}s, the stream is dead"
                ) from None
            yield msg
    finally:
        aclose = getattr(it, "aclose", None)
        if aclose is not None:
            try:
                await aclose()
            except Exception:
                pass

Three details in there earn their place.

Driving the iterator by hand with it.__anext__() is what makes per-message timing possible. A plain async for gives you no seam to put a clock in.

asyncio.wait_for cancels the awaitable it was given when the timeout fires, and the docs are explicit that it “waits for aw to be cancelled” before raising, so the total wait can exceed your timeout slightly. That is fine here. What matters is that the pending read does not keep running behind your back.

The finally block closes the underlying iterator. Python’s docs are blunt about skipping this: if an async generator exits early through an exception, its cleanup code can run “during the event loop shutdown when the async-generator garbage collection hook is called,” and the caller is told to explicitly call aclose() to detach it from the loop. You are always exiting this one early, because that is the entire point.

Step 2: retry the attempt, not the stream

A dead stream cannot be resumed by consuming it again. You need a fresh one, which means the retry has to sit around the code that opens the stream.

The other job of this layer is to name the failure. Empty output is a dead stream, and it is worth saying so out loud, because the next thing that happens to empty output is that it gets handed to a parser, and the parser reports a parse error. My old code raised “no JSON found in agent response,” which sent me to the wrong half of the system for weeks.

# runner.py
"""Bounded retry around a whole generation attempt."""
from __future__ import annotations

import asyncio
import logging

LOG = logging.getLogger(__name__)


class EmptyGeneration(RuntimeError):
    """The stream ended without producing any output."""


async def collect(stream, timeout_s: float) -> str:
    """Drain a watchdog-wrapped stream into one string."""
    from watchdog import iter_with_idle_timeout

    chunks = []
    async for msg in iter_with_idle_timeout(stream, timeout_s):
        chunks.append(msg)
    text = "".join(chunks)
    if not text.strip():
        raise EmptyGeneration(
            "generator produced no output (0 chars): the stream died before "
            "emitting anything. This is not a parsing problem."
        )
    return text


async def generate_with_retries(
    open_stream, *, timeout_s: float = 120.0, attempts: int = 3, delay_s: float = 20.0
) -> str:
    """Call `open_stream()` up to `attempts` times, watchdogged each time.

    `open_stream` is a zero-argument callable returning a fresh async
    iterable. It must be a callable, not a stream, because a dead stream
    cannot be re-consumed.
    """
    for attempt in range(1, attempts + 1):
        try:
            return await collect(open_stream(), timeout_s)
        except Exception as e:
            LOG.warning("attempt %d/%d failed: %s", attempt, attempts, e)
            if attempt == attempts:
                raise
            await asyncio.sleep(delay_s)
    raise AssertionError("unreachable")

Note the bound. Three attempts, then it gives up and raises. Google’s SRE book is direct about this: “Limit retries per request. Don’t retry a given request indefinitely.” An unbounded retry against a model API that is failing because it is overloaded is a way to make the overload worse, and the chapter walks through exactly how that amplification compounds. Three attempts twenty seconds apart still finishes well inside my job’s window.

For a real call, open_stream is the callable that starts your model request. With the Anthropic Python SDK that is the client.messages.stream(...) context manager; with an agent SDK it is whatever returns the message iterator. The watchdog does not care what is producing the messages.

Use it, then verify it

Point it at three fake streams: one healthy, one that hangs forever, and one that hangs on its first call and works on its second.

# demo.py
"""Three fake streams, one runner: healthy, hangs, then recovers."""
from __future__ import annotations

import asyncio
import logging

from runner import generate_with_retries

logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(message)s")


async def healthy():
    for word in ["Sleep ", "was ", "short. ", "Run ", "easy."]:
        await asyncio.sleep(0.05)
        yield word


async def hangs():
    yield "Sleep "
    await asyncio.sleep(3600)  # never delivers another message
    yield "never gets here"


def make_stream_factory():
    """First call hangs, second call is healthy. Mimics a transient death."""
    calls = {"n": 0}

    def factory():
        calls["n"] += 1
        return hangs() if calls["n"] == 1 else healthy()

    return factory


async def main():
    text = await generate_with_retries(healthy, timeout_s=1.0, attempts=3, delay_s=0.1)
    print(f"healthy stream  -> {text!r}")

    flaky = make_stream_factory()
    text = await generate_with_retries(flaky, timeout_s=1.0, attempts=3, delay_s=0.1)
    print(f"recovered       -> {text!r}")

    try:
        await generate_with_retries(hangs, timeout_s=1.0, attempts=2, delay_s=0.1)
    except Exception as e:
        print(f"gave up         -> {type(e).__name__}: {e}")


asyncio.run(main())

Put the three files in one directory and run it. On Python 3.14 that produced:

WARNING attempt 1/3 failed: no stream message for 1s, the stream is dead
WARNING attempt 1/2 failed: no stream message for 1s, the stream is dead
WARNING attempt 2/2 failed: no stream message for 1s, the stream is dead
healthy stream  -> 'Sleep was short. Run easy.'
recovered       -> 'Sleep was short. Run easy.'
gave up         -> StreamIdleTimeout: no stream message for 1s, the stream is dead

The warnings go to stderr and the results to stdout, so your terminal may interleave them differently. The whole run took under four seconds, which is the real check: without the watchdog, the first hanging stream would still be sitting there an hour later.

Gotchas

The exception you catch is not asyncio.TimeoutError on modern Python. As of 3.11, asyncio.wait_for “raises TimeoutError instead of asyncio.TimeoutError”. The two names are now aliases so both happen to work, but if you write except asyncio.TimeoutError out of habit you are relying on an alias rather than the documented behavior. Catch the builtin.

Passing a stream instead of a factory quietly breaks the retry. This is the easy version of the mistake to make, because the signature still typechecks and the happy path still passes. Attempt two re-consumes the already-exhausted iterator, gets nothing, and you get an empty result reported as a generation failure. Symptom: retries that never succeed and always fail identically and instantly. The escape is in the signature above, open_stream is a callable, and your test for it is the flaky-factory case, not the healthy one.

Do not cache a failure. This release also added a disk cache in front of a separate model call that renders a coaching line, after the review counted nine identical calls on one day of repeated report renders. A cache in front of a flaky call is a trap if you write the result unconditionally: one dead stream pins its fallback text until something invalidates it. Cache on success only, and key on a hash of the fully-built prompt so any input change regenerates.

A silent failure needs a loud channel. The watchdog and the retry make the job fail correctly, which still left me with a job that failed quietly. Failure now fires a distinct desktop notification and exits non-zero, and the scheduler runs a second no-op-if-already-done pass hours later. If your only success signal is a thing appearing, then absence is your only failure signal, and absence is easy to miss for weeks. Ask me how I know.

Sources

• Streaming messages, Anthropic API docs — the SSE event flow, ping events, and the documented capture-and-resume recovery strategy for interrupted streams.
• asyncio task docs, wait_for, Python — timeout cancellation semantics and the 3.11 change from asyncio.TimeoutError to TimeoutError.
• Python language reference, asynchronous generator methods — aclose() and why a caller must explicitly finalize a generator it exits early.
• Addressing Cascading Failures, Google SRE book — bounding retries per request, retry budgets, and how retry volume amplifies an overload.

Changelog

• release: 0.23.0 — facet-review loop: brief resilience, plan-vs-actual charts, plan-tool + prompt fixes (dev → main) (#117) (6df9c87)