Test AutoGen Termination Conditions

AutoGen termination condition testing treats each stop rule as a stateful async callable over controlled message or event deltas. Test non-match and match transitions, message, token, and timeout boundaries, AND or OR composition, reset behavior, and custom StopMessage output without a live model. Then add one focused team test to prove the condition is wired into execution.

The current AutoGen termination guide defines built-in conditions and their composition model. Conditions receive the delta sequence produced since their previous evaluation and return a StopMessage or None. They carry state during a run and are reset by the team after run() or run_stream() completes.

That contract is narrower than end-to-end multi-agent quality. Testing multi-agent systems covers coordination, delegation, shared state, and outcomes. Here the question is simpler: given an exact sequence of events, does the configured rule stop at the intended boundary and explain why?

What does AutoGen termination condition testing prove?

A termination condition is part of application control flow. It decides when a team has done enough work, exceeded a limit, requested an external action, reached a named source, or emitted an agreed signal. If that decision is wrong, a run may continue after completion, stop before required work, or end for a fallback cap while reporting a misleading status.

Condition-level tests prove five things. First, the rule ignores input that should not stop the run. Second, it recognizes the exact boundary that should stop it. Third, it accumulates state correctly across delta calls. Fourth, it returns a useful StopMessage. Fifth, reset restores the instance for a new run. None of these needs an agent or model.

Keep condition tests beside broader safeguards without merging their assertions. Testing AI agents can verify required output, policy, retrieval, and task behavior, while the condition test verifies only the stop transition. Evaluating an AI agent's tool use can determine whether tool execution was appropriate before completion. A run can satisfy its termination rule after poor work, or produce good work and fail to stop. Separate results make both defects visible.

Define the operational status associated with each stop reason. Approval or a structured completion event may represent normal completion. Timeout, token exhaustion, and maximum messages may represent an incomplete bounded run. External termination may represent cancellation. Tests should assert the status mapping used by callers so a common TaskResult container does not erase why execution ended.

The repository scenario ai-agents-autogen-termination-scenario in AI_AGENTS_CHALLENGES, located at src/db/seed-data/challenges/expansion-ai-agents.ts, demonstrates the central failure pattern with an older API: exact equality misses a sentinel surrounded by prose, while a loose substring can stop on an unrelated mention. The current AgentChat classes differ from that legacy predicate interface, but the testing lesson remains valid. Feed real message shapes into the stop contract and test both missed and premature matches.

Do not infer successful termination from a bounded run alone. A maximum-message guard can stop a broken conversation, but it is not proof that the intended approval or completion condition matched. Capture and assert the stop reason. Agent task completion from execution traces addresses the larger question of whether the task was completed; a condition test proves only the stopping mechanism.

How does AutoGen evaluate a termination condition?

AgentChat conditions are asynchronous callables. The team passes a sequence of new BaseAgentEvent or BaseChatMessage items after an agent response. The condition evaluates that delta, updates any internal counters or flags, and returns either None or a StopMessage. A stateful condition must not be treated as a stateless predicate over the complete transcript unless its documentation says so.

Delta semantics affect fixtures. If MaxMessageTermination receives one message and later receives two new messages, its accumulated count reflects those calls. Passing the entire transcript on every call can double-count and produce a test that does not resemble team execution. Name fixture batches delta_1, delta_2, and delta_3 to make this visible.

Once a condition is terminated, the base contract may reject another call until reset. Test this behavior rather than continuing to feed messages and accepting any result. An independent case should create a fresh instance or await reset(). Shared module-level condition objects make order-dependent tests because one case can leave terminal state for another.

Composition wraps the same lifecycle. left | right stops when either child stops. left & right requires both. The composed object must be reset as a unit before reuse. A table should cover which child becomes true first, whether the other child has already accumulated state, and what stop message the composition exposes.

The AutoGen condition reference is the source for constructor parameters and current classes. Pin AutoGen in the project, write examples against that version, and review API changes during upgrades instead of carrying assumptions from the 0.2 ConversableAgent interface.

Which message and event fixtures should tests create?

Build the smallest object that carries the signal under test. TextMentionTermination needs text content and may need a source. SourceMatchTermination needs messages from named sources. TokenUsageTermination needs reported usage. A custom function-call condition may need a ToolCallExecutionEvent. A stop-on-handoff condition needs the relevant handoff message.

Use fixture names that describe meaning: assistant_draft, critic_approval, runner_completion_without_sentinel, sentinel_mentioned_as_data, and completion_with_sentinel. This is clearer than message1 and helps reviewers see missing cases. Keep content short enough that the triggering token or source is obvious.

Include adversarially similar cases. For a text condition, test different case only if the condition is expected to be case-sensitive or normalized by your wrapper. Test punctuation, surrounding prose, sentinel as quoted data, sentinel from the wrong source, empty content, and non-text events. Do not claim behavior that the built-in class does not promise; adapt the application contract or add a custom condition when stricter matching is required.

For token usage, include no usage, usage below each configured limit, exact threshold, and a delta that crosses the threshold. If prompt and completion limits are configured separately, test them separately before testing a total. The application may need a fallback because not every model client reports usage on every event.

Fixture builders reduce noise, but do not hide fields that control the result. A builder for TextMessage should require source and content explicitly. If usage is optional, make the test pass it by name. This gives a reviewer direct evidence of why the condition should or should not stop.

Build a second fixture dimension for event batching. A team can emit several inner events in one response, while the termination condition receives the delta associated with that response. Test a matching item at the start, middle, and end of a multi-item delta when order could affect the custom rule. For counters, verify whether agent events are included according to constructor options instead of assuming every event increments a message limit.

Preserve message source as part of the oracle. The same text from an assistant, critic, tool runner, or user proxy can carry different authority. A completion sentinel copied by the wrong participant should not stop a source-restricted policy. This is a control-flow permission, even when the condition API expresses it as a source filter.

How do you test message, text, and source conditions?

Begin with built-in conditions that do not depend on time or model usage. The following async tests construct TextMessage objects and call conditions directly. They prove delta accumulation, text matching, source filtering, terminal state, and reset without creating a team.

import pytest

from autogen_agentchat.conditions import (
    MaxMessageTermination,
    SourceMatchTermination,
    TextMentionTermination,
)
from autogen_agentchat.messages import TextMessage


def text(content: str, source: str = "assistant") -> TextMessage:
    return TextMessage(content=content, source=source)


@pytest.mark.asyncio
async def test_max_message_termination_counts_deltas():
    condition = MaxMessageTermination(max_messages=3)

    assert await condition([text("draft")]) is None
    assert await condition([text("review", "critic")]) is None

    stop = await condition([text("approved", "critic")])
    assert stop is not None
    assert condition.terminated is True

    await condition.reset()
    assert condition.terminated is False
    assert await condition([text("fresh run")]) is None


@pytest.mark.asyncio
async def test_text_mention_can_be_limited_to_critic():
    condition = TextMentionTermination("APPROVE", sources=["critic"])

    assert await condition([text("APPROVE", "assistant")]) is None
    stop = await condition([text("APPROVE", "critic")])

    assert stop is not None


@pytest.mark.asyncio
async def test_source_match_stops_after_runner_response():
    condition = SourceMatchTermination(["runner"])

    assert await condition([text("working", "assistant")]) is None
    assert await condition([text("tests complete", "runner")]) is not None

The text test intentionally asks only whether the built-in mention condition sees the text from an allowed source. It does not add end-anchoring that the built-in rule does not advertise. If the application contract says a sentinel must appear alone on the final line, create a custom condition and test that exact grammar.

Source restrictions prevent another participant from ending a run merely by quoting the sentinel. They do not prove that the authorized source's message is correct. Multi-agent handoff and delegation boundaries addresses whether roles and handoffs behave correctly beyond this local stop decision.

How do you test token and timeout limits deterministically?

Token conditions rely on usage attached to messages or events. The test must construct reported usage; word count is not a substitute for provider token accounting. Use RequestUsage from the current AutoGen model interfaces and attach it through the message field supported by the pinned version. Then assert below-threshold and threshold-crossing deltas.

import pytest

from autogen_agentchat.conditions import (
    TimeoutTermination,
    TokenUsageTermination,
)
from autogen_agentchat.messages import TextMessage
from autogen_core.models import RequestUsage


def metered(content: str, prompt: int, completion: int) -> TextMessage:
    return TextMessage(
        content=content,
        source="assistant",
        models_usage=RequestUsage(
            prompt_tokens=prompt,
            completion_tokens=completion,
        ),
    )


@pytest.mark.asyncio
async def test_total_token_limit_crosses_on_second_delta():
    condition = TokenUsageTermination(max_total_token=100)

    assert await condition([metered("draft", 40, 10)]) is None
    stop = await condition([metered("revision", 35, 15)])

    assert stop is not None
    assert condition.terminated is True


@pytest.mark.asyncio
async def test_timeout_crosses_a_controlled_monotonic_boundary(
    monkeypatch,
):
    clock = iter([100.0, 104.9, 105.0, 200.0])
    monkeypatch.setattr(
        "autogen_agentchat.conditions._terminations.time.monotonic",
        lambda: next(clock),
    )
    condition = TimeoutTermination(timeout_seconds=5)

    assert await condition([]) is None
    stop = await condition([])

    assert stop is not None
    assert stop.source == "TimeoutTermination"
    assert condition.terminated is True

    await condition.reset()
    assert condition.terminated is False

If the current installed package uses a renamed usage field or type, update the fixture to the pinned reference rather than inventing a conversion. Also test a message with no usage. The expected behavior should follow the official implementation, while application safety can add an independent message limit or timeout.

Timeout tests are vulnerable to CI scheduling noise when they assert exact elapsed milliseconds. The current stable implementation constructs and resets TimeoutTermination from time.monotonic(), so the example patches that module boundary while using the documented constructor, call, terminated, and reset APIs. The iterator supplies construction, before-threshold, threshold, and reset timestamps. Keep this condition-level test pinned because its clock lookup is an implementation detail; a package upgrade should trigger review if that lookup moves.

At the team layer, timeout means the stop rule observed its duration, not necessarily that all underlying work was cancelled. Test cancellation and cleanup separately if they matter. Agent token budget and tail-latency scenarios covers broader operational decisions; the condition suite should remain focused on boundary behavior.

How do AND and OR termination conditions behave?

OR composition is common for "goal reached or budget exhausted." Build each child first, then combine them. Test a run stopped by the goal before the cap and a run stopped by the cap without the goal. The stop reason must distinguish those outcomes because operations may treat completion differently from budget exhaustion.

AND composition represents "both requirements observed." For example, a reviewer may need to approve and a runner may need to report completion. Feed the signals in both orders. Confirm the first signal is retained while the condition waits for the second. Then reset and repeat in reverse order.

Do not test only the final truth table. Inspect the intermediate non-terminal calls, terminal message, child states where exposed, and reset. Composition bugs often appear in order and reuse rather than the obvious final combination.

Measuring agent step efficiency can use stop reasons as trace context, but it should not reward a run merely for ending early. A termination test protects the rule; an evaluation decides whether that rule produced a good task outcome.

How do you test a custom TerminationCondition?

Create a custom condition when the application needs a rule not represented by built-ins, such as an end-anchored approval, a structured result status, or a specific executed function. The class should track terminal state, return None before the match, return a StopMessage on the match, implement reset, and follow the base contract after termination.

This example stops only when the critic's final non-empty line is exactly APPROVED. A mention inside explanatory prose does not match.

from collections.abc import Sequence

from autogen_agentchat.base import TerminatedException, TerminationCondition
from autogen_agentchat.messages import (
    BaseAgentEvent,
    BaseChatMessage,
    StopMessage,
    TextMessage,
)


class FinalLineApproval(TerminationCondition):
    def __init__(self) -> None:
        self._terminated = False

    @property
    def terminated(self) -> bool:
        return self._terminated

    async def __call__(
        self,
        messages: Sequence[BaseAgentEvent | BaseChatMessage],
    ) -> StopMessage | None:
        if self._terminated:
            raise TerminatedException("Condition already reached")

        for message in messages:
            if not isinstance(message, TextMessage):
                continue
            if not isinstance(message.content, str):
                continue
            lines = [line.strip() for line in message.content.splitlines()]
            final_line = next((line for line in reversed(lines) if line), "")
            if message.source == "critic" and final_line == "APPROVED":
                self._terminated = True
                return StopMessage(
                    content="Critic supplied final approval",
                    source="final_line_approval",
                )
        return None

    async def reset(self) -> None:
        self._terminated = False

Test quoted approval, approval from the wrong source, approval followed by more text, empty content, non-text events, correct final approval, second call after termination, and reset. If the class needs serialization through AutoGen's component system, implement and test the configuration contract shown in the official base reference rather than omitting it silently.

Custom logic should remain narrow. Do not embed broad task evaluation in a stop predicate. A complex quality rubric belongs in an evaluator or reviewer agent whose structured result can then trigger a simple condition.

What reset and reuse cases can break a test suite?

The most common local mistake is a condition instance shared across tests. One test reaches the limit, another starts with terminal state, and order determines the result. Create conditions inside each test or use a function-scoped fixture. Do not use a session-scoped condition unless shared state is the behavior under test.

Direct calls require explicit reset when the same object represents a new logical run. Assert both terminated becoming false and counters returning to their initial behavior. For a maximum-message condition, one message after reset should act like the first message, not the next message after the previous run.

Team execution has its own lifecycle. The official guide states that the team resets its condition after a run completes. Add one wiring test that runs the same deterministic team twice and confirms each run can reach the intended stop independently. That test proves team ownership of reset; it should not replace direct boundary cases.

Composition adds another reuse risk. Reset the combined condition, not one child selected by the test. Then prove signals from the previous run do not satisfy the new run. For custom conditions, clear every accumulated field, not only the top-level boolean.

Concurrency is a separate contract. Do not share one stateful condition object between simultaneous teams unless AutoGen and the application explicitly support that ownership model. Give each run its own instance. Multi-agent observability architecture can record run and stop identities so accidental cross-run state is visible.

How do you run the termination test workflow in pytest?

Use this workflow for each condition or composed stop policy:

1. Write the stopping rule in terms of observable messages, events, source, usage, or time.
2. Choose the built-in condition or define the smallest custom rule that matches it.
3. Build typed delta fixtures, including non-matches that resemble the terminal signal.
4. Create a fresh condition and assert its initial state.
5. Feed one or more non-terminal deltas and assert None.
6. Cross the exact boundary and inspect the returned StopMessage.
7. Assert behavior after termination according to the base contract.
8. Await reset and prove the same instance behaves like a fresh condition.
9. Test AND or OR composition in every meaningful signal order.
10. Add one deterministic team test that distinguishes intended termination from a fallback cap.

Record the stop source and content in integration evidence. If the team ends because of a maximum-message guard, report that as budget exhaustion, not task completion. If an external stop button ends it, report external cancellation. These distinctions support agent trajectory evaluation and incident diagnosis.

Avoid full transcript equality. A condition test should assert the exact fixture fields needed for the rule. A team test can assert stop reason, bounded trajectory properties, and required structured output. Exact assistant prose is usually outside the stop contract.

The team wiring case should use controlled agents or replayed outputs where the current AutoGen test utilities support them. Arrange one run that reaches the intended condition and one that reaches a safety cap instead. Assert that their terminal reasons differ, even if both return a TaskResult. This protects dashboards and callers from reporting an exhausted run as successful completion.

Also test cancellation ownership around external termination when the application exposes a stop action. Setting an external condition communicates that the team should stop; it does not by itself prove that a downstream process, tool call, or background task released resources. Keep cleanup assertions with the component that owns those resources and correlate its evidence with the team's stop reason.

For production diagnostics, record the configured condition type, relevant limit, run ID, terminal source, and sanitized terminal content. Avoid recording complete transcripts when a small event slice explains the decision. A reviewer should be able to tell whether the run ended by approval, source match, timeout, token budget, message cap, or external request without reconstructing control flow from final prose.

FAQ

Are AutoGen termination conditions stateful?

Yes. Many conditions accumulate counts, usage, elapsed time, or match state across delta calls. Test intermediate calls and the terminal transition. Create a fresh instance per independent test or reset deliberately so earlier messages cannot affect a later logical run.

Does AutoGen reset a condition after a run?

The AgentChat team runtime resets its condition after a completed run or stream according to the official guide. Direct calls made by a unit test do not run that team lifecycle, so the test should await reset() before treating the same instance as a new run.

Can termination conditions be combined?

Yes. OR stops when either child reaches its rule, while AND waits for both. Test each child, both signal orders, the combined stop reason, and reset. Do not assume that a truth-table assertion covers accumulated child state.

How do token conditions work when usage is missing?

They can only count usage that the message or event reports. Include a missing-usage fixture and follow the pinned implementation's documented result. If the application requires a stop even without usage reporting, combine token limits with a separate message or timeout condition.

How can timeout tests avoid flakiness?

Keep them free of model and network work, use a controlled clock or synchronization boundary where possible, and assert before-expiry versus after-expiry behavior. At the team layer, inspect the stop reason instead of relying on a narrow elapsed-time assertion on shared CI.

What should a custom condition return?

Return None while the run should continue and a StopMessage when the rule matches. The message should identify the reason and source. Also expose terminal state, handle calls after termination according to the base contract, and implement a complete async reset.

Conclusion: Treat every stop reason as tested control flow

Termination is not a decorative team setting. It is executable control flow with state, ordering, composition, and lifecycle behavior. Direct async tests can prove those mechanics without paying for or waiting on a model, while one focused team case proves the condition is connected to the runtime.

Build the message table first, then cover the exact boundary, stop message, reset, and composed alternatives. When traces distinguish approval, completion, timeout, token limit, and message cap, a stopped run becomes evidence rather than an ambiguous absence of further output.