Execution Graph Tracing

Pipelex captures pipeline executions as directed graphs for visualization and debugging. The pipelex/graph/ module provides tracing infrastructure, a canonical data model (GraphSpec), and multiple rendering backends (Mermaid, ReactFlow).

Design Principle

Pipeline executions form implicit graphs: pipes call other pipes, data flows between them. The graph module makes this structure explicit:

1. Trace at runtime: Instrument pipe execution to capture nodes (pipes) and edges (relationships)
2. Store canonically: GraphSpec is a versioned, renderer-agnostic JSON model
3. Render as needed: Transform GraphSpec into Mermaid diagrams or ReactFlow visualizations

Pipe Execution → GraphTracer → GraphSpec → Renderers → HTML/Mermaid

Usage Variants

Interfaces

CLI Commands

# Run pipeline and generate graph
pipelex run pipe my_pipe --graph

# Include full serialized data
pipelex run pipe my_pipe --graph --graph-full-data

# Exclude data (previews only)
pipelex run pipe my_pipe --graph --graph-no-data

# Dry run with graph tracing
pipelex run pipe my_pipe --dry-run --graph --mock-inputs

API

from pipelex.pipeline.runner import PipelexMTHDSProtocol
from pipelex.system.pipe_run_mode import PipeRunMode

# Execute with graph tracing via config
runner = PipelexMTHDSProtocol(
    execution_config=config.with_execution_overrides(generate_graph=True),
)
response = await runner.execute(
    pipe_code="my_pipe",
)
pipe_output = response.pipe_output

# Dry run with graph: the same runner in DRY mode with mock inputs — no separate code path.
dry_runner = PipelexMTHDSProtocol(
    pipe_run_mode=PipeRunMode.DRY,
    execution_config=config.with_execution_overrides(generate_graph=True, mock_inputs=True),
)
response = await dry_runner.execute(pipe_code="my_pipe")
graph_spec = response.pipe_output.graph_spec

Outputs

The three companions are the validation report's pipe_io_contracts, input_form and output_form under the standard's names: a graph carries every payload the run produced and nothing that says what those payloads are, and a graph viewer (@pipelex/mthds-ui's GraphViewer) shows a data node's value only when it holds both the contracts and the output form for the producing pipe_ref. They are built once, at the end of PipeRun.run inside the run's own library window (the builders need the loaded library, which PipelineRunner.execute tears down before the CLI writes anything), over the run library's own pipes (the validated bundle and whatever library_dirs loaded beside it; a dependency package's pipes are keyed by alias in the library and described by no blueprint the host holds, so they are left out, as a validation of the bundle leaves them out), and carried on pipe_output.pipe_io_artifacts beside graph_spec; a failure of the builders, whatever its kind, is reported on pipe_output.pipe_io_artifacts_error, mirroring graph_assembly_error, and never fails the run: the build sits in the run's finally ahead of the delivery, so an exception escaping it would replace the run's outcome and skip the delivery of a run that completed. The build is gated by describe_pipe_io on the trace context, which the run setup derives from graphspec_json, and which validate's own graph dry run turns off since validate already built the artifacts for its report. Every writer that serializes the graphspec writes them beside it: save_graph_outputs_to_dir for the CLIs, the delivery executor for a hosted results prefix. They follow the graphspec_json inclusion flag, so a run that asks for no graphspec builds nothing that describes one, and a written graphspec owns the three names beside it: when the run carries no artifacts, a companion left by an earlier run in the same directory is removed, with a warning naming it, rather than paired with the new graph. The agent CLI, which always writes a graphspec with --graph, applies that inclusion before the run so the companions are built in the window rather than lost to a render-time override. Their bytes are those of the projection fixture corpus for the same bundle, since both go through render_pipe_io_artifact_files (pipelex/core/pipes/pipe_io_artifacts.py). The pipe_io_artifacts field crosses the SPI payload as pipe_io_artifacts_dump; the synchronous /execute response carries the description once the API relays that field, and a hosted run's results prefix once our Temporal plugin builds the artifacts in the crate's window.

Architecture

flowchart TB
    subgraph EXECUTION["Pipeline Execution"]
        direction TB
        PIPE["PipeAbstract.run_pipe()"]
        JOB["JobMetadata"]
        PIPE --> JOB
    end

    subgraph TRACING["Graph Tracing"]
        direction TB
        MGR["GraphTracerManager<br/>(singleton)"]
        TRACER["GraphTracer"]
        CTX["TraceContext"]
        MGR --> TRACER
        TRACER --> CTX
    end

    subgraph MODEL["Canonical Model"]
        direction TB
        SPEC["GraphSpec"]
        NODE["NodeSpec"]
        EDGE["EdgeSpec"]
        IO["IOSpec"]
        SPEC --> NODE
        SPEC --> EDGE
        NODE --> IO
    end

    subgraph ANALYSIS["Pre-computed Analysis"]
        direction TB
        GA["GraphAnalysis"]
        TREE["containment_tree"]
        STUFF["stuff_registry"]
        GA --> TREE
        GA --> STUFF
    end

    subgraph RENDER["Renderers"]
        direction TB
        MF["MermaidflowFactory"]
        RF["ReactFlow HTML Generator"]
        MF --> HTML1["Mermaid HTML"]
        RF --> HTML2["ReactFlow HTML"]
    end

    JOB --> CTX
    CTX --> TRACER
    TRACER --> SPEC
    SPEC --> GA
    GA --> MF
    GA --> RF

Core Components

GraphSpec

The canonical, versioned data model for execution graphs. Designed for JSON serialization and renderer-agnostic storage.

class GraphSpec(BaseModel):
    graph_id: str
    created_at: datetime
    pipeline_ref: PipelineRef
    nodes: list[NodeSpec]
    edges: list[EdgeSpec]
    usage: GraphUsageSpec | None
    meta: dict[str, Any]

    def to_json(self) -> str:
        return self.model_dump_json(by_alias=True, indent=2)

usage (and its per-node counterpart NodeSpec.usage) carries the run's inference usage attributed to graph position — see Per-node Usage Attribution for the three usage states and the invariants a consumer must branch on.

meta.format is always "mthds" on newly emitted Pipelex graphs. meta.mode records provenance for renderers and shared tooling: Pipelex execution graphs emit "dry" for dry-run/mock execution and "live" for real execution. The shared renderer also accepts "static" for graphs produced by a static MTHDS graph builder.

Node Types

Edge Types

Every EdgeSpec also carries an optional boolean (default false). On a DATA edge it marks that the producer's output is declared optional (? presence marker) — the value flowed on this run but may be absent on others. Renderers can use it to draw the edge distinctly (e.g. dashed).

Node Status

A SKIPPED node also carries skip_reason (a human-readable sentence naming the absent input, e.g. skipped because input 'source' is absent). A lifted pipe with a plural output still writes a real empty-list value, so its node registers that output and downstream DATA edges resolve normally; a lifted singular output is a recorded absence with no output spec.

Implementation

Tracing Flow

# 1. Manager opens tracer for pipeline run
manager = GraphTracerManager.get_or_create_instance()
trace_context = manager.open_tracer(
    graph_id=pipeline_run_id,
    data_inclusion=config.data_inclusion,
    pipeline_ref_domain="my_domain",
    pipeline_ref_main_pipe="my_pipe",
)

# 2. Context flows through JobMetadata to each pipe
job_metadata = JobMetadata(
    pipeline_run_id=pipeline_run_id,
    trace_context=trace_context,
)

# 3. Each pipe reports start/end to tracer
node_id, child_context = manager.on_pipe_start(
    trace_context=trace_context,
    pipe_code="extract_text",
    pipe_type="PipeExtract",
    node_kind=NodeKind.OPERATOR,
    started_at=datetime.now(timezone.utc),
    input_specs=[...],
)

# 4. On completion, report success with output
manager.on_pipe_end_success(
    lookup_key=trace_context.lookup_key,
    node_id=node_id,
    ended_at=datetime.now(timezone.utc),
    output_spec=IOSpec(...),
)

# 5. Manager closes tracer and returns GraphSpec
graph_spec = manager.close_tracer(pipeline_run_id)

Event-Log Transport and the Scoped Override

Trace events travel through an EventLogProtocol backend (pipelex/tracing/): the tracer emits events into it during the run (write side, wired in pipeline_run_setup), and assemble_tracing reads them back after the run to build the GraphSpec and usage aggregates (read side, triggered from PipeRun.run). Both sides normally build their backend instance independently from tracing_config via make_event_log — NDJSON files or DynamoDB bridge the two instances through external storage.

The assembled usage rides back on pipe_output.tokens_usages (with any assembly failure on usage_assembly_error), which the sync /execute response returns directly. For delivery-enabled runs (a storage target set), the delivery executor (pipelex/pipe_run/delivery_executor.py) also persists it as a tokens_usages.json result artifact — {"tokens_usages": [...], "usage_assembly_error": null} — next to working_memory.json, the main_stuff.* renders, and the graph outputs (graphspec.json with its three companions pipe_io_contracts.json, input_form.json and output_form.json, see Outputs), so a durable client polling result files gets the same usage records a sync caller does. The artifact is written unconditionally on every successful result delivery: explicit nulls mean usage assembly was off for that run. A failed run stores no result files at all, so an absent tokens_usages.json means either the run failed or it was delivered before the artifact existed — tell those apart from the delivery status, not from the file's presence.

For fully in-process runs, pipelex.runtime_hub.scoped_event_log pins one shared instance for both sides instead:

from pipelex.runtime_hub import scoped_event_log
from pipelex.tracing.in_memory_event_log import InMemoryEventLog

with scoped_event_log(InMemoryEventLog()):
    response = await runner.execute(...)  # graph assembles in memory

Semantics:

• The override is ContextVar-scoped (mirrors scoped_pipe_router), so concurrent runs with separate scopes never cross-contaminate, and the prior value is restored on exit.
• A set override implies tracing-enabled: it is honored even when runtime.tracing.is_enabled is False, on both the write side and the read side's early-return.
• Lifecycle: the read side does not close() the scoped instance and the machinery never calls cleanup() on it — but the write-side tracer DOES call close() on its event log at teardown, before the read side assembles. A scoped event log's close() must therefore be idempotent or a no-op (as InMemoryEventLog's is); scoping a backend whose close() releases a real resource would break its own assembly read.

This is what lets a graph-producing dry-run trace entirely in memory (no NDJSON file, no DynamoDB round-trip). Both dry-run entrypoints rely on it: dry_run_pipe_in_process (pipelex/pipe_run/dry_run_in_process.py — the graph arm of protocol validate and of the single Temporal validation activity) and dry_run_pipeline (pipelex/pipeline/dry_run_pipeline.py) itself — these functions exist to produce a graph, so they install their own scoped InMemoryEventLog rather than depending on the host having tracing configured (a host with runtime.tracing.is_enabled = false, like pipelex-api's /validate in direct mode, still gets its graph). A run that nonetheless finishes without a graph raises the typed DryRunGraphNotProducedError.

TraceContext Propagation

TraceContext is a serializable context that flows through pipe execution:

class TraceContext(BaseModel):
    graph_id: str  # Unique graph identifier
    parent_node_id: str | None  # Parent pipe's node ID
    node_sequence: int  # Counter for generating node IDs
    data_inclusion: DataInclusionConfig  # What data to capture

    def copy_for_child(self, child_node_id: str, *, next_sequence: int) -> TraceContext:
        """Create context for nested pipe execution."""
        return TraceContext(
            graph_id=self.graph_id,
            parent_node_id=child_node_id,
            node_sequence=next_sequence,
            data_inclusion=self.data_inclusion,
        )

Producer/Consumer Paradigm

Data flow in the graph follows a producer/consumer model:

• Producers: Nodes that create or output data (stuff). When a pipe outputs a value, it becomes the producer of that data item.
• Consumers: Nodes that receive or use data as input. When a pipe takes a value as input, it becomes a consumer of that data item.

Each piece of data is identified by a digest (a unique hash). This allows the tracer to track where data originates and where it flows, even when the same value passes through multiple pipes.

Producer Node ──(outputs)──► Stuff (digest: abc123) ──(inputs)──► Consumer Node

This paradigm enables:

• Automatic DATA edge generation without explicit wiring
• Visualization of data lineage across the execution graph
• Debugging by tracing which pipe produced unexpected output

Data Flow Edge Generation

DATA edges are generated at teardown by correlating input/output digests:

def _generate_data_edges(self) -> None:
    for consumer_node_id, node_data in self._nodes.items():
        for input_spec in node_data.input_specs:
            if input_spec.digest is None:
                continue
            producer_node_id = self._stuff_producer_map.get(input_spec.digest)
            if producer_node_id and producer_node_id != consumer_node_id:
                self.add_edge(
                    source_node_id=producer_node_id,
                    target_node_id=consumer_node_id,
                    edge_kind=EdgeKind.DATA,
                    label=input_spec.name,
                )

GraphAnalysis

Pre-computed analysis layer that extracts common information for renderers:

analysis = GraphAnalysis.from_graphspec(graph_spec)

# Lookups
node = analysis.nodes_by_id["node_123"]
children = analysis.get_children("controller_node")
is_root = analysis.is_root("node_123")

# Data flow
stuff_info = analysis.get_stuff_info(digest="abc123")
producer = analysis.get_producer(digest="abc123")
consumers = analysis.get_consumers(digest="abc123")

Rendering

Mermaidflow

Converts GraphSpec to Mermaid flowchart syntax with controller subgraphs:

from pipelex.graph.mermaidflow.mermaidflow_factory import MermaidflowFactory

mermaidflow = MermaidflowFactory.make_from_graphspec(
    graph_spec,
    graph_config=graph_config,
    direction=FlowchartDirection.TOP_DOWN,
    include_subgraphs=True,
)

print(mermaidflow.mermaid_code)

Output structure:

• Controllers rendered as subgraphs
• Operators rendered as rectangles inside subgraphs
• Stuff nodes (data items) rendered as stadium shapes
• DATA edges connect producers → stuff → consumers

In the standalone viewer (mermaidflow.html), clicking a stuff node opens its JSON content, and an image or PDF output also offers a preview rendered from the URL that JSON carries.

ReactFlow HTML

ReactFlow HTML is generated directly from GraphSpec — no intermediate ViewSpec layer. The HTML generator embeds GraphSpec as JSON and the client-side JavaScript handles dataflow analysis, layout, and rendering.

Configuration

GraphConfig

# pipelex.toml (default values)
[interpreter.pipeline_execution.graph]

[interpreter.pipeline_execution.graph.data_inclusion]
stuff_json_content = true       # Include full serialized data
error_stack_traces = true       # Include full stack traces
pipe_and_concept_registry = true  # Include pipe and concept registries in the GraphSpec

[interpreter.pipeline_execution.graph.graphs_inclusion]
graphspec_json = true           # Generate GraphSpec JSON, and its three companions beside it
mermaidflow_mmd = true          # Generate Mermaid code
mermaidflow_html = true         # Generate Mermaid HTML
reactflow_html = true           # Generate ReactFlow HTML

MermaidRenderingConfig

ReactFlowRenderingConfig

IOSpec Data Capture

IOSpec captures input/output data for nodes:

class IOSpec(BaseModel):
    name: str  # Variable name
    concept: str | None  # Concept code
    content_type: str | None  # MIME type
    preview: str | None  # Truncated preview (max 200 chars)
    size: int | None  # Content size
    digest: str | None  # Unique identifier for data flow
    data: str | dict[str, Any] | list[str] | list[dict[str, Any]] | None  # Full serialized content
    extra: dict[str, Any]  # Extra markers, e.g. `optional` set on a declared-optional (`?`) output

Validation

GraphSpec validation enforces invariants:

from pipelex.graph.validation import validate_graphspec

validate_graphspec(graph_spec)

Syntax Quick Reference

Files Reference

Next Steps

• Architecture Overview
• Image Handling in LLM Prompts