State and Routing¶

ScenarioState is the mutable runtime object that accumulates data as a scenario executes. Edge conditions query this state via dot-path expressions to determine which node to visit next. The edge resolution algorithm evaluates conditions in definition order, taking the first match.

This page explains the structure of ScenarioState, how the engine updates it after each turn, and which patterns cover common routing needs. For defining the graph structure itself, see Building Scenarios. For asserting on the completed execution, see Outcome Criteria.

1. What It Is¶

ScenarioState is the mutable runtime object that accumulates data as a scenario executes. It is created at the start of execution and updated after every turn. After each turn completes, the engine writes the turn's verify_result, the parsed template fields, the updated visit count for the current node, and a structured entry in node_results. Edge conditions query this object using dot-path strings to determine which node to visit next.

State is not inspected manually during normal use. You interact with it indirectly: by writing dot-path conditions in add_edge() calls and by reading final_state from the ScenarioExecutionResult returned by the runner.

2. Core Idea¶

State accumulates automatically; you define conditions declaratively.

After each turn, the engine updates:

• verify_result: the boolean template verification result from the latest turn (or None if the template did not run)
• parsed: the dictionary of parsed template fields from the latest turn
• node_visits[node_id]: incremented by one for the current node
• node_results[node_id]: a structured dict containing verify_result, parsed (template fields), and rubric (trait scores) for that node (last-write-wins on revisits)

Edge conditions query these using dot-path strings:

Path string	Resolved value
"verify_result"	Latest turn's template verification result
"parsed.drug"	parsed["drug"] from the latest turn
"node_visits.retry"	Number of times the retry node has been visited
"node_results.ask.verify_result"	Template result from the most recent ask turn

The first conditional edge whose condition matches the current state wins. If no conditional edge matches, the first unconditional edge is used as a fallback. If there are no edges at all, execution terminates.

3. Anatomy¶

ScenarioState is a dataclass with the following fields:

Field	Type	Description
turn	int	Current turn number (0-indexed)
current_node	str	Node being executed
verify_result	bool \| None	Template verification result from the latest turn
parsed	dict[str, Any]	Parsed template fields from the latest turn
node_visits	dict[str, int]	Visit count per node
history	list[TurnRecord]	All previous turn records
accumulated	dict[str, Any]	Custom state populated via state_update callbacks
node_results	dict[str, dict]	Auto-populated per-node results (verify_result, parsed, rubric)

TurnRecord captures one turn's full output. Its fields are described in the Reference section.

4. How It Works¶

Each turn follows this sequence (matches karenina/scenario/manager.py):

1. Set state.current_node to the current node id.
2. Execute the node: run the answering model, then the verification pipeline.
3. Build a TurnRecord capturing the question, trace, raw response, parsed answer, and turn-level verify_result.
4. Write state.node_results[current_node] with a dict containing verify_result, parsed (template fields), and rubric (trait scores). Last write wins on revisits.
5. If the node has a state_update callback, call it with the current accumulated dict and parsed fields. The engine snapshots accumulated first; if the callback raises, the snapshot is restored.
6. Increment state.turn by 1.
7. Increment state.node_visits[current_node] by 1.
8. Append the TurnRecord to state.history.
9. Write state.verify_result and state.parsed from this turn (these reflect the latest turn for downstream edge conditions).

After updating state, the engine calls resolve_next_node with the outbound edges from the current node and the updated state. The function returns a (target, edge) tuple; the runner uses both the target node id and the matched edge (e.g., to inspect a handover strategy when crossing an agent_identity boundary).

Edge resolution algorithm:

1. Collect all edges where source == current_node.
2. Iterate in definition order.
3. Save the first unconditional edge (no condition and no condition_callable) as the fallback (its target plus the edge object itself).
4. For each conditional edge, evaluate its condition against the current state. Return (edge.target, edge) for the first matching edge.
5. If no conditional edge matched, return the saved fallback (target, edge).
6. If no fallback exists, return (None, None) (implicit terminal; execution ends).

In [2]:

# Dot-path conditions passed to add_edge() become StateCheck objects internally.
# The engine evaluates them in the order the edges were added.

scenario = Scenario("drug-identification")

scenario.add_edge("ask", "retry", when={"verify_result": False})
scenario.add_edge("ask", "deep_dive", when={"parsed.confidence": "high"})
scenario.add_edge("retry", END, when={"node_visits.retry": 3})

# Simulate state after a failed turn on "ask"
state = ScenarioState(
    turn=0,
    current_node="ask",
    verify_result=False,
    parsed={"confidence": "low"},
    node_visits={"ask": 1},
    history=[],
    accumulated={},
    node_results={},
)

ask_edges = scenario.edges_from("ask")
next_node, _ = resolve_next_node(ask_edges, state)
print(f"Next node after failed ask: {next_node}")

# Simulate state after a successful turn with high confidence
state2 = ScenarioState(
    turn=1,
    current_node="ask",
    verify_result=True,
    parsed={"confidence": "high"},
    node_visits={"ask": 2},
    history=[],
    accumulated={},
    node_results={},
)

next_node2, _ = resolve_next_node(ask_edges, state2)
print(f"Next node after successful ask with high confidence: {next_node2}")

# Dot-path conditions passed to add_edge() become StateCheck objects internally. # The engine evaluates them in the order the edges were added. scenario = Scenario("drug-identification") scenario.add_edge("ask", "retry", when={"verify_result": False}) scenario.add_edge("ask", "deep_dive", when={"parsed.confidence": "high"}) scenario.add_edge("retry", END, when={"node_visits.retry": 3}) # Simulate state after a failed turn on "ask" state = ScenarioState( turn=0, current_node="ask", verify_result=False, parsed={"confidence": "low"}, node_visits={"ask": 1}, history=[], accumulated={}, node_results={}, ) ask_edges = scenario.edges_from("ask") next_node, _ = resolve_next_node(ask_edges, state) print(f"Next node after failed ask: {next_node}") # Simulate state after a successful turn with high confidence state2 = ScenarioState( turn=1, current_node="ask", verify_result=True, parsed={"confidence": "high"}, node_visits={"ask": 2}, history=[], accumulated={}, node_results={}, ) next_node2, _ = resolve_next_node(ask_edges, state2) print(f"Next node after successful ask with high confidence: {next_node2}")

Next node after failed ask: retry
Next node after successful ask with high confidence: deep_dive

5. Patterns¶

a. Simple routing on verify_result¶

Route to a retry node if the template failed, or to an end node if it passed:

In [3]:

scenario_a = Scenario("simple-routing")

scenario_a.add_edge("ask", "retry", when={"verify_result": False})
scenario_a.add_edge("ask", END)  # unconditional fallback: reached only if verify_result is True

scenario_a.add_edge("retry", END)

state_pass = ScenarioState(
    turn=0,
    current_node="ask",
    verify_result=True,
    parsed={},
    node_visits={"ask": 1},
    history=[],
    accumulated={},
    node_results={},
)
state_fail = ScenarioState(
    turn=0,
    current_node="ask",
    verify_result=False,
    parsed={},
    node_visits={"ask": 1},
    history=[],
    accumulated={},
    node_results={},
)

print(f"Pass path: {resolve_next_node(scenario_a.edges_from('ask'), state_pass)[0]}")
print(f"Fail path: {resolve_next_node(scenario_a.edges_from('ask'), state_fail)[0]}")

scenario_a = Scenario("simple-routing") scenario_a.add_edge("ask", "retry", when={"verify_result": False}) scenario_a.add_edge("ask", END) # unconditional fallback: reached only if verify_result is True scenario_a.add_edge("retry", END) state_pass = ScenarioState( turn=0, current_node="ask", verify_result=True, parsed={}, node_visits={"ask": 1}, history=[], accumulated={}, node_results={}, ) state_fail = ScenarioState( turn=0, current_node="ask", verify_result=False, parsed={}, node_visits={"ask": 1}, history=[], accumulated={}, node_results={}, ) print(f"Pass path: {resolve_next_node(scenario_a.edges_from('ask'), state_pass)[0]}") print(f"Fail path: {resolve_next_node(scenario_a.edges_from('ask'), state_fail)[0]}")

Pass path: __end__
Fail path: retry

b. Routing on parsed fields¶

Route based on a field extracted by the answer template. Here the template parses a confidence field from the response, and the scenario branches on its value:

In [4]:

scenario_b = Scenario("parsed-routing")

scenario_b.add_edge("ask", "clarify", when={"parsed.confidence": "low"})
scenario_b.add_edge("ask", END)  # fallback for any other confidence value

state_low = ScenarioState(
    turn=0,
    current_node="ask",
    verify_result=True,
    parsed={"confidence": "low"},
    node_visits={"ask": 1},
    history=[],
    accumulated={},
    node_results={},
)
state_high = ScenarioState(
    turn=0,
    current_node="ask",
    verify_result=True,
    parsed={"confidence": "high"},
    node_visits={"ask": 1},
    history=[],
    accumulated={},
    node_results={},
)

print(f"Low confidence path: {resolve_next_node(scenario_b.edges_from('ask'), state_low)[0]}")
print(f"High confidence path: {resolve_next_node(scenario_b.edges_from('ask'), state_high)[0]}")

scenario_b = Scenario("parsed-routing") scenario_b.add_edge("ask", "clarify", when={"parsed.confidence": "low"}) scenario_b.add_edge("ask", END) # fallback for any other confidence value state_low = ScenarioState( turn=0, current_node="ask", verify_result=True, parsed={"confidence": "low"}, node_visits={"ask": 1}, history=[], accumulated={}, node_results={}, ) state_high = ScenarioState( turn=0, current_node="ask", verify_result=True, parsed={"confidence": "high"}, node_visits={"ask": 1}, history=[], accumulated={}, node_results={}, ) print(f"Low confidence path: {resolve_next_node(scenario_b.edges_from('ask'), state_low)[0]}") print(f"High confidence path: {resolve_next_node(scenario_b.edges_from('ask'), state_high)[0]}")

Low confidence path: clarify
High confidence path: __end__

c. Custom accumulated state via state_update¶

ScenarioNode accepts an optional state_update callable. After each turn on that node, the engine calls state_update(accumulated, parsed_fields) and replaces accumulated with the returned dict. This allows tracking values across revisits, such as a running count.

Before calling the callback, the engine takes a deepcopy snapshot of accumulated. If the callback raises an exception, the snapshot is restored and the error is logged at warning level. This guarantees that a buggy state_update cannot corrupt accumulated state for subsequent turns.

In [5]:

# In real usage, state_update is attached to a ScenarioNode.
# Here we demonstrate the accumulation pattern directly.


def track_attempts(accumulated: dict, parsed_fields: dict) -> dict:
    """Increment attempt counter on each visit."""
    return {**accumulated, "attempts": accumulated.get("attempts", 0) + 1}


# Simulate three visits
acc: dict = {}
for _ in range(3):
    acc = track_attempts(acc, {})

print(f"Accumulated after 3 visits: {acc}")

# An edge can then route on accumulated.attempts:
scenario_c = Scenario("accumulated-routing")
scenario_c.add_edge("probe", END, when={"accumulated.attempts": 3})
scenario_c.add_edge("probe", "probe")  # fallback: loop back

state_3 = ScenarioState(
    turn=2,
    current_node="probe",
    verify_result=None,
    parsed={},
    node_visits={"probe": 3},
    history=[],
    accumulated={"attempts": 3},
    node_results={},
)
print(f"At 3 attempts, next node: {resolve_next_node(scenario_c.edges_from('probe'), state_3)[0]}")

# In real usage, state_update is attached to a ScenarioNode. # Here we demonstrate the accumulation pattern directly. def track_attempts(accumulated: dict, parsed_fields: dict) -> dict: """Increment attempt counter on each visit.""" return {**accumulated, "attempts": accumulated.get("attempts", 0) + 1} # Simulate three visits acc: dict = {} for _ in range(3): acc = track_attempts(acc, {}) print(f"Accumulated after 3 visits: {acc}") # An edge can then route on accumulated.attempts: scenario_c = Scenario("accumulated-routing") scenario_c.add_edge("probe", END, when={"accumulated.attempts": 3}) scenario_c.add_edge("probe", "probe") # fallback: loop back state_3 = ScenarioState( turn=2, current_node="probe", verify_result=None, parsed={}, node_visits={"probe": 3}, history=[], accumulated={"attempts": 3}, node_results={}, ) print(f"At 3 attempts, next node: {resolve_next_node(scenario_c.edges_from('probe'), state_3)[0]}")

Accumulated after 3 visits: {'attempts': 3}
At 3 attempts, next node: __end__

d. Cross-node result lookups via node_results¶

node_results lets edges on a later node reference results from an earlier node. This is useful in multi-branch scenarios where a final synthesis node needs to know how an earlier node resolved:

In [6]:

scenario_d = Scenario("cross-node-routing")

# The "synthesize" node routes based on whether the earlier "ask" node passed.
scenario_d.add_edge("synthesize", "explain", when={"node_results.ask.verify_result": False})
scenario_d.add_edge("synthesize", END)

state_ask_failed = ScenarioState(
    turn=2,
    current_node="synthesize",
    verify_result=True,
    parsed={},
    node_visits={"ask": 1, "synthesize": 1},
    history=[],
    accumulated={},
    node_results={
        "ask": {"verify_result": False, "parsed": {}, "rubric": {}},
    },
)
state_ask_passed = ScenarioState(
    turn=2,
    current_node="synthesize",
    verify_result=True,
    parsed={},
    node_visits={"ask": 1, "synthesize": 1},
    history=[],
    accumulated={},
    node_results={
        "ask": {"verify_result": True, "parsed": {}, "rubric": {}},
    },
)

print(f"ask failed: next is {resolve_next_node(scenario_d.edges_from('synthesize'), state_ask_failed)[0]}")
print(f"ask passed: next is {resolve_next_node(scenario_d.edges_from('synthesize'), state_ask_passed)[0]}")

scenario_d = Scenario("cross-node-routing") # The "synthesize" node routes based on whether the earlier "ask" node passed. scenario_d.add_edge("synthesize", "explain", when={"node_results.ask.verify_result": False}) scenario_d.add_edge("synthesize", END) state_ask_failed = ScenarioState( turn=2, current_node="synthesize", verify_result=True, parsed={}, node_visits={"ask": 1, "synthesize": 1}, history=[], accumulated={}, node_results={ "ask": {"verify_result": False, "parsed": {}, "rubric": {}}, }, ) state_ask_passed = ScenarioState( turn=2, current_node="synthesize", verify_result=True, parsed={}, node_visits={"ask": 1, "synthesize": 1}, history=[], accumulated={}, node_results={ "ask": {"verify_result": True, "parsed": {}, "rubric": {}}, }, ) print(f"ask failed: next is {resolve_next_node(scenario_d.edges_from('synthesize'), state_ask_failed)[0]}") print(f"ask passed: next is {resolve_next_node(scenario_d.edges_from('synthesize'), state_ask_passed)[0]}")

ask failed: next is explain
ask passed: next is __end__

6. Reference¶

Dot-path resolution¶

Path	Resolves to
"verify_result"	state.verify_result
"turn"	state.turn
"current_node"	state.current_node
"parsed.<field>"	state.parsed.get(field)
"accumulated.<field>"	state.accumulated.get(field)
"node_visits.<node_id>"	state.node_visits.get(node_id, 0)
"node_results.<node>"	state.node_results.get(node, {})
"node_results.<node>.verify_result"	state.node_results[node]["verify_result"]
"node_results.<node>.parsed.<field>"	state.node_results[node]["parsed"][field]
"node_results.<node>.rubric.<trait>"	state.node_results[node]["rubric"][trait]

Missing keys return None, except node_visits.<node_id>, which returns 0.

TurnRecord fields¶

Field	Type	Description
node_id	str	The node that produced this turn
question_text	str	The rendered question text sent to the model
raw_response	str	The model's raw response text
parsed_fields	dict[str, Any]	Template fields parsed from the response
verify_result	bool \| None	Template verification result for this turn

ScenarioExecutionResult fields¶

Field	Type	Description
scenario_id	str	Identifier for the scenario
status	"completed" \| "limit_reached" \| "error" \| "timeout"	How execution terminated. ScenarioManager only ever emits "completed", "limit_reached", or "error". "timeout" is a reserved schema value that the manager does not currently produce.
path	list[str]	Ordered list of node ids visited
turn_count	int	Total number of turns executed
history	list[TurnRecord]	Full turn history
turn_results	list[VerificationResult]	Per-turn verification result objects
final_state	ScenarioState	State as of the last turn
outcome_results	dict[str, bool \| int \| float]	Evaluated outcome criteria, keyed by name
terminal_failure	ScenarioTerminalFailure \| None	Populated only when status="error", otherwise None. Carries the structured reason the run stopped early.

When a scenario stops with status="error", terminal_failure holds a ScenarioTerminalFailure describing why. It is a small dataclass exported from karenina.schemas.scenario with four fields:

Field	Type	Description
node_id	str	The node whose turn raised the terminal failure
category	str	Failure category (the classified failure kind)
stage	str	Pipeline stage where the failure surfaced
reason	str	Human-readable explanation of the stop

A "completed" or "limit_reached" run leaves terminal_failure as None, so checking it is the reliable way to distinguish a real error from a normal stop.

Turn limit¶

VerificationConfig.scenario_turn_limit (default: 20) sets the maximum number of turns before the runner terminates the scenario with status="limit_reached". Set this field in your VerificationConfig to override the default.

Evaluation mode¶

Scenarios auto-detect evaluation mode per turn based on whether a rubric is present (per-question rubric or global rubric). Setting evaluation_mode='rubric_only' on VerificationConfig has no effect in scenarios: the ScenarioManager ignores it and emits a UserWarning explaining why. If you need rubric-only evaluation, attach rubrics to your questions without answer templates and the auto-detection will select template_and_rubric or template_only as appropriate.

state_update snapshot and rollback¶

A state_update callback attached to a ScenarioNode can mutate the accumulated dict freely. The runtime guards against buggy callbacks with a deepcopy snapshot and rollback:

1. Before invoking state_update(accumulated, parsed_fields), the runtime takes a copy.deepcopy(accumulated) snapshot.
2. If the callback returns normally, the returned dict replaces the prior accumulated.
3. If the callback raises any exception, the snapshot is restored, the exception is logged at warning level (with traceback), and execution continues at the next turn with the pre-callback accumulated state.

This guarantee means a callback that fails on one turn cannot corrupt the accumulated state seen by subsequent turns, and never aborts the scenario by itself; downstream edge conditions on accumulated.<field> will read consistent values regardless of whether the callback succeeded.

condition_callable failure semantics¶

Edge condition_callable lambdas (the form passed as when=lambda state: ...) are wrapped in a try/except in _edge_matches. If the callable raises any exception, the runtime logs at warning level (with traceback) and treats the edge as unmatched (returns False). The scenario continues; resolution falls through to the next conditional edge or the unconditional fallback. A buggy callable cannot crash a run, but it can silently mis-route execution: prefer declarative StateCheck when={...} forms when you can express the condition that way.

Public API¶

evaluate_state_check and resolve_next_node are re-exported from karenina.scenario and form the public surface of edge resolution.

Symbol	Purpose
karenina.scenario.evaluate_state_check(check, state)	Evaluate a single StateCheck against a ScenarioState. Resolves the dot-path field and applies check.verify_with.check(value, expected). Returns bool.
karenina.scenario.resolve_next_node(edges, state)	Choose the next node given outbound edges. Returns (target_node_id, matched_edge). The matched_edge is needed by the runner to inspect handover settings on the followed edge (see Handover). Returns (None, None) when no edge matches and there is no fallback (implicit terminal).

Both helpers are useful in tests and when prototyping conditions outside a full scenario run.

7. Next Steps¶

• Sycophancy Tutorial: end-to-end walkthrough of a sycophancy resistance scenario that uses state-driven routing
• Handover: per-edge context routing fired when a matched edge carries a handover setting; documents TaggedMessage, transcript strategies, and callable handovers
• Execution: the ScenarioExecutor and ScenarioExecutorConfig that host scenario runs, parallel/sequential modes, replicates, and SchemaOrg checkpoint persistence
• Scenario Internals: contributor-level detail on the execution engine is in the karenina-guide skill reference at references/advanced/scenario-internals.md, not a separate docs page
• Building Scenarios: constructing the graph, adding nodes and edges, serialization