Flow and Count Assertions

Assert on tool-calling order, sequences, counts, and arguments.

Overview

Flow and count assertions verify the behavior of an agent — which tools it called, in what order, how many times, and with what arguments.

Assertions reference

assert_called

Tool was called at least once:

s.assert_called("search")

assert_never_called

Tool was never called:

s.assert_never_called("delete_account")

assert_called_before

Tool a was called before tool b (positional ordering across all turns):

s.assert_called_before("assess_risk", "make_decision")

assert_called_times

Tool was called between min and max times:

s.assert_called_times("search", min=1, max=3)

# Exactly once
s.assert_called_times("assess_risk", min=1, max=1)

assert_called_with

Tool was called with specific argument values (partial match):

s.assert_called_with("search", query="Q3 revenue")

Matches if any call to the tool includes the specified arguments. Extra arguments are allowed. If no call matches, the error message shows the closest mismatch.

assert_tool_succeeded

Tool was called and every execution succeeded (no error in any result):

s.assert_tool_succeeded("search")

assert_max_turns

Trace has at most n turns:

s.assert_max_turns(5)

assert_total_duration_under

Total trace duration is under a threshold (milliseconds):

s.assert_total_duration_under(ms=10_000)

Uses wall clock time for active sessions.

Flow patterns

assert_flow matches the ordered sequence of tool calls against a pattern. Patterns support ... (Ellipsis) for “any number of calls here”:

# Exact consecutive match — first two calls must be search then summarize
s.assert_flow(["search", "summarize"])

# Allow any calls between search and summarize
s.assert_flow(["search", ..., "summarize"])

# Unanchored — match anywhere in the trace
s.assert_flow([..., "search", ..., "summarize", ...])

How anchoring works

Patterns are anchored to start and end by default:

Pattern	Matches	Doesn’t match
["a", "b"]	a → b	a → b → c (extra at end)
["a", ..., "b"]	a → x → y → b	x → a → b (extra at start)
[..., "a", "b"]	x → a → b	a → b → x (extra at end)
[..., "a", ..., "b", ...]	anything containing a then b	b → a (wrong order)

Full example

def test_agent_behavior(tmp_path):
    with reagent_flow.session("vendor-approval", trace_dir=str(tmp_path)) as s:
        s.log_llm_call(
            tool_calls=[{"name": "extract_vendor_packet", "arguments": {"vendor": "ClearVoice AI"}}],
        )
        s.log_tool_result("extract_vendor_packet", result={"vendor_name": "ClearVoice AI"})

        s.log_llm_call(
            tool_calls=[{"name": "review_security", "arguments": {"vendor": "ClearVoice AI"}}],
        )
        s.log_tool_result("review_security", result={"risk": "HIGH"})

        s.log_llm_call(
            tool_calls=[{"name": "record_decision", "arguments": {"decision": "ESCALATE"}}],
        )
        s.log_tool_result("record_decision", result={"decision": "ESCALATE"})

    # Flow assertions
    s.assert_called("review_security")
    s.assert_never_called("delete_data")
    s.assert_called_before("review_security", "record_decision")
    s.assert_called_times("review_security", min=1, max=1)
    s.assert_called_with("record_decision", decision="ESCALATE")
    s.assert_flow(["extract_vendor_packet", "review_security", "record_decision"])
    s.assert_max_turns(5)