> ## Documentation Index
> Fetch the complete documentation index at: https://reagent-ai.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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:

```python theme={null}
s.assert_called("search")
```

### assert\_never\_called

Tool was never called:

```python theme={null}
s.assert_never_called("delete_account")
```

### assert\_called\_before

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

```python theme={null}
s.assert_called_before("assess_risk", "make_decision")
```

### assert\_called\_times

Tool was called between `min` and `max` times:

```python theme={null}
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):

```python theme={null}
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):

```python theme={null}
s.assert_tool_succeeded("search")
```

### assert\_max\_turns

Trace has at most `n` turns:

```python theme={null}
s.assert_max_turns(5)
```

### assert\_total\_duration\_under

Total trace duration is under a threshold (milliseconds):

```python theme={null}
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":

```python theme={null}
# 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

```python theme={null}
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)
```