core.graph_execution_helpers

Pure helper functions for prompt resolution, abort checking, and failure propagation during graph (DAG) execution.

These functions operate on results_by_name (the executor’s accumulated result dict) rather than the shared prompt_attr_history, making them suitable for level-by-level graph execution where results are only written to the shared history after the entire graph completes.

Pure helper functions for prompt resolution, abort checking, and failure propagation during graph (DAG) execution.

These functions operate on results_by_name (the executor’s accumulated result dict) rather than the shared prompt_attr_history, making them suitable for level-by-level graph execution where results are only written to the shared history after the entire graph completes.

build_graph_history_dict(results_by_name)

Build a {name: response_text} mapping from executor results.

Parameters:

results_by_name (dict[str, dict[str, Any]]) – Dict mapping prompt_name to result dicts with response, status, etc.

Returns:

Dict mapping prompt_name to response text (str).

Return type:

dict[str, str]

resolve_graph_prompt(prompt_spec, results_by_name)

Resolve interpolation and history injection for a graph node.

Performs two-phase prompt assembly:

1. Interpolation: Resolves {{name.response}} patterns using results from earlier DAG levels.

2. History injection: For each name in prompt_spec["history"], formats the prior Q&A as <conversation_history> XML. Entries already interpolated are deduplicated.

Parameters:

• prompt_spec (dict[str, Any]) – The prompt dict with keys prompt, history, etc.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (resolved_prompt, set_of_interpolated_names).

Return type:

tuple[str, set[str]]

check_abort_condition(prompt_spec, results_by_name)

Check if a prompt’s abort_condition triggers a cascade abort.

Parameters:

• prompt_spec (dict[str, Any]) – Prompt dict with optional abort_condition key.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (should_abort, trace_or_None, error_or_None).

Return type:

tuple[bool, str | None, str | None]

should_skip_for_failed_deps(node, results_by_name, nodes)

Check if a node should be skipped because any dependency failed.

Only failed dependencies trigger a skip. Skipped dependencies do NOT cascade — a node whose only dependency was skipped will still execute (its condition will determine whether it should proceed).

Parameters:

• node (PromptNode) – The prompt node to check.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results indexed by prompt_name.

• nodes (dict[int, PromptNode]) – All graph nodes indexed by sequence number.

Returns:

Tuple of (should_skip, reason_string).

Return type:

tuple[bool, str]

Functions

build_graph_history_dict(results_by_name)

Build a {name: response_text} mapping from executor results.

Parameters:

results_by_name (dict[str, dict[str, Any]]) – Dict mapping prompt_name to result dicts with response, status, etc.

Returns:

Dict mapping prompt_name to response text (str).

Return type:

dict[str, str]

check_abort_condition(prompt_spec, results_by_name)

Check if a prompt’s abort_condition triggers a cascade abort.

Parameters:

• prompt_spec (dict[str, Any]) – Prompt dict with optional abort_condition key.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (should_abort, trace_or_None, error_or_None).

Return type:

tuple[bool, str | None, str | None]

resolve_graph_prompt(prompt_spec, results_by_name)

Resolve interpolation and history injection for a graph node.

Performs two-phase prompt assembly:

1. Interpolation: Resolves {{name.response}} patterns using results from earlier DAG levels.

2. History injection: For each name in prompt_spec["history"], formats the prior Q&A as <conversation_history> XML. Entries already interpolated are deduplicated.

Parameters:

• prompt_spec (dict[str, Any]) – The prompt dict with keys prompt, history, etc.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results from prior DAG levels.

Returns:

Tuple of (resolved_prompt, set_of_interpolated_names).

Return type:

tuple[str, set[str]]

should_skip_for_failed_deps(node, results_by_name, nodes)

Check if a node should be skipped because any dependency failed.

Only failed dependencies trigger a skip. Skipped dependencies do NOT cascade — a node whose only dependency was skipped will still execute (its condition will determine whether it should proceed).

Parameters:

• node (PromptNode) – The prompt node to check.

• results_by_name (dict[str, dict[str, Any]]) – Accumulated results indexed by prompt_name.

• nodes (dict[int, PromptNode]) – All graph nodes indexed by sequence number.

Returns:

Tuple of (should_skip, reason_string).

Return type:

tuple[bool, str]