Check Stop Reasons

From Handling to a Policy

Lesson 20 introduced every stop reason and how to handle each one in isolation. In a real application, stop-reason handling needs to be:

• centralized: one place decides what each stop reason means for your app,
• bounded: automatic recovery, such as continuing or retrying, must have limits so it can never loop forever,
• observable: you need to know how often each stop reason occurs, because a rising rate of truncations or refusals is an early warning sign, and
• testable: the logic should be verified without making real API calls.

This lesson turns stop-reason checks into a policy: a defined, tested set of actions with limits and metrics.

A Stop-Reason Policy Table

Start by writing down what your application does for each stop reason. This table becomes the specification for your code:

Why limits matter: each automatic action costs a request. Without limits, a request that keeps hitting max_tokens or pause_turn could loop and spend tokens indefinitely.

Classifying Outcomes

Callers rarely need to know the raw stop reason. They need to know what to do with the result. Map stop reasons to a small set of outcomes:

from enum import Enum


class Outcome(str, Enum):
    COMPLETE = "complete"         # Safe to use as a final answer
    TRUNCATED = "truncated"       # Usable with a warning, or must be redone
    DECLINED = "declined"         # Do not show as an answer
    NEEDS_TOOLS = "needs_tools"   # The tool loop must continue
    INCOMPLETE = "incomplete"     # A server-side operation did not finish
    EMPTY = "empty"               # No usable content

Outcome(str, Enum) makes each member also a string, so outcomes can be logged and stored as plain text.

Measuring Stop Reasons

Count every stop reason your application receives:

from collections import Counter

stop_reason_counts: Counter[str] = Counter()
stop_reason_counts[message.stop_reason] += 1

Then watch the rates, not just the counts:

Log the request ID with every non-end_turn outcome (Lesson 29), so individual cases can be investigated.

Stop Reasons in Other Contexts

The same values appear outside messages.create():

• Streaming: the stop reason arrives near the end of the stream, and the final message object from the stream helpers includes it (Lessons 53–55).
• Message Batches: each successful batch result contains a full message with its own stop reason (Lesson 137). A refusal inside a batch is still a "succeeded" result at the batch level.

Your policy should apply in all three places, which is another reason to keep it in one function.

Practical Example: A Bounded Stop-Reason Policy

This module implements the policy table: it recovers automatically within limits, classifies the final outcome, and records metrics. A fake client is used to test every branch without real API calls.

"""stop_policy.py: a bounded, observable stop-reason policy."""

from collections import Counter
from dataclasses import dataclass
from enum import Enum


class Outcome(str, Enum):
    COMPLETE = "complete"
    TRUNCATED = "truncated"
    DECLINED = "declined"
    NEEDS_TOOLS = "needs_tools"
    INCOMPLETE = "incomplete"
    EMPTY = "empty"


@dataclass
class PolicyResult:
    outcome: Outcome
    message: object          # The final Message
    attempts: int


FALLBACK_MODEL = "claude-opus-4-8"   # Check the "Refusals and fallback" docs for current guidance
MAX_PAUSE_CONTINUATIONS = 3
metrics: Counter[str] = Counter()


def text_of(message) -> str:
    return "".join(b.text for b in message.content if b.type == "text").strip()


def run_with_policy(client, **request) -> PolicyResult:
    request = dict(request)
    request["messages"] = list(request["messages"])
    attempts = 0
    escalated = refused_once = nudged = False
    pauses = 0

    while True:
        attempts += 1
        message = client.messages.create(**request)
        reason = message.stop_reason
        metrics[reason] += 1

        if reason in ("end_turn", "stop_sequence"):
            if text_of(message) or reason == "stop_sequence":
                return PolicyResult(Outcome.COMPLETE, message, attempts)
            if nudged:
                return PolicyResult(Outcome.EMPTY, message, attempts)
            nudged = True
            request["messages"] += [{"role": "assistant", "content": message.content},
                                    {"role": "user", "content": "Please continue."}]

        elif reason == "max_tokens":
            if escalated:
                return PolicyResult(Outcome.TRUNCATED, message, attempts)
            escalated = True
            request["max_tokens"] = request["max_tokens"] * 4

        elif reason == "model_context_window_exceeded":
            return PolicyResult(Outcome.TRUNCATED, message, attempts)

        elif reason == "refusal":
            if refused_once:
                return PolicyResult(Outcome.DECLINED, message, attempts)
            refused_once = True
            request["model"] = FALLBACK_MODEL

        elif reason == "pause_turn":
            if pauses >= MAX_PAUSE_CONTINUATIONS:
                return PolicyResult(Outcome.INCOMPLETE, message, attempts)
            pauses += 1
            request["messages"] += [{"role": "assistant", "content": message.content}]

        elif reason == "tool_use":
            return PolicyResult(Outcome.NEEDS_TOOLS, message, attempts)

        else:
            metrics["unknown"] += 1
            return PolicyResult(Outcome.INCOMPLETE, message, attempts)


# ---------- Offline tests with a fake client ----------

from types import SimpleNamespace


def fake_message(stop_reason: str, text: str = ""):
    content = [SimpleNamespace(type="text", text=text)] if text else []
    return SimpleNamespace(stop_reason=stop_reason, content=content)


class FakeClient:
    """Returns pre-defined responses in order and records each request."""

    def __init__(self, responses):
        self.responses = list(responses)
        self.requests = []
        self.messages = SimpleNamespace(create=self._create)

    def _create(self, **request):
        self.requests.append(request)
        return self.responses.pop(0)


def test_policy() -> None:
    base = {"model": "claude-opus-5-5", "max_tokens": 1024,
            "messages": [{"role": "user", "content": "Hi"}]}

    client = FakeClient([fake_message("max_tokens", "partial"), fake_message("end_turn", "full")])
    result = run_with_policy(client, **base)
    assert result.outcome is Outcome.COMPLETE and result.attempts == 2
    assert client.requests[1]["max_tokens"] == 4096

    client = FakeClient([fake_message("refusal"), fake_message("refusal")])
    result = run_with_policy(client, **base)
    assert result.outcome is Outcome.DECLINED
    assert client.requests[1]["model"] == FALLBACK_MODEL

    client = FakeClient([fake_message("pause_turn")] * 4)
    result = run_with_policy(client, **base)
    assert result.outcome is Outcome.INCOMPLETE and result.attempts == 4

    client = FakeClient([fake_message("end_turn"), fake_message("end_turn")])
    result = run_with_policy(client, **base)
    assert result.outcome is Outcome.EMPTY

    assert base["messages"] == [{"role": "user", "content": "Hi"}]   # Caller's data untouched
    print("All policy tests passed.")
    print("Metrics:", dict(metrics))


if __name__ == "__main__":
    test_policy()

Output:

All policy tests passed.
Metrics: {'max_tokens': 1, 'end_turn': 3, 'refusal': 2, 'pause_turn': 4}

How It Works

1. Copies first: request = dict(request) and list(request["messages"]) create copies, so the policy can add messages and change limits without modifying the caller's data. The final assertion in the tests verifies this.
2. One loop, explicit state: flags such as escalated, refused_once, and nudged, plus the pauses counter, record which recoveries have been used. Each recovery can happen only up to its limit, so the loop always ends.
3. Each branch implements one row of the policy table. For example, max_tokens multiplies the limit by 4 once, then returns TRUNCATED. refusal switches to the fallback model once, then returns DECLINED.
4. tool_use is handed off, not handled. The tool loop is a separate responsibility (Lessons 85–92), and mixing it in here would make both harder to test.
5. Unknown stop reasons are counted and returned as INCOMPLETE rather than crashing, which keeps the application safe when the API adds new values.
6. metrics counts every stop reason received, including those from automatic retries. In production, send these counts to your monitoring system.
7. FakeClient imitates client.messages.create() using SimpleNamespace objects, which have only the attributes the policy reads. Each test supplies a sequence of responses and checks the outcome, the number of attempts, and the requests that were sent, all without network access or cost.

Common Mistakes

Unbounded Automatic Recovery

What happens: a while loop retries on max_tokens or pause_turn forever, consuming tokens.

Fix: give every automatic action an explicit limit.

Scattering Stop-Reason Checks

What happens: different parts of the code treat the same stop reason differently.

Fix: route every response through one policy function.

Modifying the Caller's Request

What happens: appended continuation messages leak into the caller's conversation history.

Fix: copy the request and its message list before modifying them.

Not Measuring Stop Reasons

What happens: gradual increases in truncation or refusal go unnoticed until users complain.

Fix: count stop reasons, track their rates, and log request IDs for unusual outcomes.

Testing Only the Happy Path

What happens: retry and fallback branches are never exercised until production.

Fix: use a fake client to test every branch.

Best Practices

• Write the policy table first, then implement it.
• Centralize stop-reason handling in one tested function.
• Bound every automatic recovery with explicit limits.
• Return simple outcomes that callers can act on.
• Never modify caller data inside the policy.
• Record metrics for every stop reason, and alert on unusual rates.
• Test every branch offline with fake responses.

Exercise

1. Add a test in which the first response is stop_sequence with no text, and assert that the outcome is COMPLETE.
2. Change the max_tokens escalation so it never exceeds 32,000 tokens.
3. Add a rate(reason) function that returns the share of all recorded responses with a given stop reason, for example rate("max_tokens") == 0.1.
4. Explain why the pause_turn branch appends only an assistant message, while the empty-response branch appends both an assistant and a user message.

Step 2:

request["max_tokens"] = min(request["max_tokens"] * 4, 32_000)

Step 3:

def rate(reason: str) -> float:
    total = sum(count for key, count in metrics.items() if key != "unknown")
    return metrics[reason] / total if total else 0.0

"unknown" is excluded from the total because it is an extra counter, recorded in addition to the actual stop reason value.

Step 4: for pause_turn, the documented way to continue is to send the paused response back as the assistant turn, so the API resumes that same turn. For an empty end_turn, Claude considered its turn finished, and the request would end with an assistant message, which current models reject because prefill is not supported. A new user message ("Please continue.") is needed to ask for more.