feat: pass invocation_state to edge condition calls

[yananym]

Description

Add support for edge conditions that receive invocation_state, enabling conditional routing based on runtime context (feature flags, user roles, environment config) passed during
graph invocation.

Also fixes a deadlock in _compute_ready_nodes_for_resume() where conditional edges evaluating to False would block downstream nodes from ever becoming ready on interrupt/resume
workflows.

Public API Changes

GraphBuilder.add_edge() now accepts conditions with an extended signature via the EdgeConditionWithContext Protocol:

# Legacy (still works, no changes needed)
def my_condition(state: GraphState) -> bool:
    return len(state.completed_nodes) > 0

# New: receives invocation_state for runtime routing decisions
def my_condition(state: GraphState, *, invocation_state: dict, **kwargs) -> bool:
    return invocation_state.get("role") == "admin"

builder.add_edge("router", "admin_path", condition=my_condition)
graph("task", invocation_state={"role": "admin"})

Both signatures are supported indefinitely — no deprecation, no breaking changes.

Related Issues

Resolves #1346

Documentation PR

strands-agents/docs#847

Type of Change

New feature

Testing

How have you tested the change?

• 21 new unit tests covering protocol dispatch, invocation_state propagation, resume deadlock fix, and serialization round-trip
• 6 new integration tests covering conditional routing, backwards compat, combined conditions, diamond convergence, persistence/resume, and streaming events
• All 49 existing graph unit tests pass unchanged (backwards compatibility verified)
• 95% line+branch coverage on src/strands/multiagent/graph.py
• ruff check and mypy pass clean
• I ran hatch run prepare

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[github-actions]

Issue: inspect.signature() is called on every should_traverse() invocation with no caching. In graphs with many edges or cyclic execution, this could become a performance bottleneck since signature introspection is relatively expensive.

Suggestion: Cache the result per condition function using functools.lru_cache or a simple dict keyed by id(condition):

_condition_type_cache: dict[int, bool] = {}

def _is_context_condition(condition: EdgeCondition) -> TypeGuard[EdgeConditionWithContext]:
    cond_id = id(condition)
    if cond_id not in _condition_type_cache:
        try:
            sig = inspect.signature(condition)
            _condition_type_cache[cond_id] = "invocation_state" in sig.parameters
        except (ValueError, TypeError):
            _condition_type_cache[cond_id] = False
    return _condition_type_cache[cond_id]

The issue spec also mentions this: "Consider caching signature inspection results for performance if needed."

[github-actions]

Issue: The @runtime_checkable decorator is present but not used for dispatch — _is_context_condition uses inspect.signature() instead of isinstance(). This is the correct choice (since isinstance can't distinguish the two signatures structurally), but the @runtime_checkable decorator may mislead users into thinking they can use isinstance(cond, EdgeConditionWithContext) for the same purpose.

Suggestion: Consider adding a brief comment noting that @runtime_checkable is for type-checking ergonomics / documentation purposes only, not for runtime dispatch, since isinstance will match both legacy and new-style conditions.

[github-actions]

Issue: The PR introduces a public API change (GraphBuilder.add_edge() accepts a new condition signature, new types EdgeCondition and EdgeConditionWithContext are exported) but doesn't appear to have a needs-api-review label. Per the API bar-raising process, new public abstractions customers will use should be flagged for API review.

Suggestion: Consider adding the needs-api-review label. In particular, the API reviewer should evaluate:

• Whether EdgeConditionWithContext is the right naming (vs. e.g., ContextualEdgeCondition)
• Whether requiring **kwargs in the protocol is the right extensibility pattern vs. a versioned protocol approach
• Whether the invocation_state: dict[str, Any] type should be dict[str, Any] | None to match the caller side more closely

[github-actions]

Issue: invocation_state is serialized directly into the session payload without validation. If a user passes non-JSON-serializable objects (e.g., class instances, functions) in invocation_state, this will fail silently or raise an unclear error during serialization.

Suggestion: Consider either:

1. Documenting that invocation_state values must be JSON-serializable, or
2. Adding a validation/guard in serialize_state() that provides a clear error message if serialization fails due to non-serializable invocation_state values.

[yananym]

Addressed in 7cc0d04 — added _validate_invocation_state() that calls json.dumps() and raises a clear TypeError if the value isn't serializable. It's gated on session_manager is not None (serialization only matters when sessions persist), and also validated symmetrically on the deserialization path in deserialize_state.

[github-actions]

Issue: In the _is_node_ready_for_resume method, the logic evaluates all edge conditions to determine traversability. However, since invocation_state is restored from the serialized payload, but the edge conditions themselves are defined in code — if the graph code changes between serialize and deserialize (e.g., condition logic updated), the behavior could be subtly different on resume. This is inherent to the design but worth noting.

Also, is calling e.should_traverse() here redundant with the e.condition is None check? should_traverse already returns True when condition is None, so the filter could be simplified to just e.should_traverse(...):

traversable_edges = [
    e for e in incoming
    if e.should_traverse(self.state, invocation_state=self._current_invocation_state)
]

Suggestion: The explicit e.condition is None check is likely a micro-optimization to avoid the inspect.signature() call for unconditional edges. If caching is added to _is_context_condition, this optimization becomes unnecessary and the code can be simplified.

[yananym]

Addressed in 7cc0d04. You're right that with the WeakKeyDictionary cache now in place, the e.condition is None short-circuit is no longer necessary for performance. I kept it because it still avoids the method call + cache lookup entirely for unconditional edges, and makes the "unconditional edges always traverse" semantics explicit at the call site. Added an inline comment explaining the intent. Happy to simplify if you'd prefer the terser form.
On the serialize/deserialize behavior divergence — added a docstring to _compute_ready_nodes_for_resume acknowledging this explicitly: re-evaluation is intentional since invocation_state may differ between invocations, and condition logic changes taking effect on resume is consistent with the graph being defined in code rather than serialized.

[github-actions]

Issue: Tests using Graph.__new__(Graph) bypass __init__ and assign graph.edges as a list (e.g., graph.edges = [edge]) while the actual Graph class types it as set[GraphEdge]. This works at runtime since both are iterable, but it creates a type inconsistency and could mask issues where set semantics (deduplication, ordering) matter.

Suggestion: Consider using graph.edges = {edge} (set literal) in tests for consistency with the actual class, or extract a helper function that creates a minimally-valid Graph instance for testing without bypassing init.

[github-actions]

Assessment: Comment

Well-designed feature that cleanly extends the existing edge condition system with invocation_state context while maintaining full backwards compatibility. The protocol-based approach with inspect.signature() dispatch is a solid technical choice.

Review Categories

• Performance: inspect.signature() is called on every should_traverse() without caching — worth addressing for graphs with many conditional edges or cyclic execution patterns.
• API Surface: New public types (EdgeCondition, EdgeConditionWithContext) are exported but the PR may benefit from formal API review per the bar-raising process, particularly around naming and the protocol's type design choices.
• Serialization: invocation_state is persisted without validation of serializability — could produce confusing errors for users passing non-serializable values.
• Testing: Comprehensive test coverage (21 unit + 6 integration tests). Minor type inconsistency with test setup using lists where sets are expected.

The deadlock fix in _compute_ready_nodes_for_resume is a valuable correctness improvement alongside the main feature.

[yananym]

Issue: The PR introduces a public API change (GraphBuilder.add_edge() accepts a new condition signature, new types EdgeCondition and EdgeConditionWithContext are exported) but doesn't appear to have a needs-api-review label. Per the API bar-raising process, new public abstractions customers will use should be flagged for API review.

Suggestion: Consider adding the needs-api-review label. In particular, the API reviewer should evaluate:

I do not think i have permissions to add labels to this PR