dr-graph

dr-graph represents hashable computation graphs as data, interprets them deterministically, and provides exact flow optimization primitives. Graph structure is separate from caller-supplied node behavior and optimization.

• Definitions describe reusable graph topology, node fields, dependencies, and variable requirements.
• Configuration models concrete variable values and validates the resulting graph.
• Identity owns graph definition, configuration, and identity; every complete graph configuration gets a versioned hash that callers may nest in larger keys.
• Execution runs execute_graph as the sole serial entry point, delegating node behavior to the caller through RunNode.
• Results models per-node and graph-level outcomes, including reuse of completed node outputs when continuing execution.
• Flow optimization solves exact min-cost flow and balanced separable convex transportation problems independently of computation-graph execution.
• Infra
  • Assembly creates graphs programmatically, including deterministic namespacing and rewiring of subgraphs.
  • Core contains shared errors, field and input-source models, topology helpers, and strict-JSON validation.

The following sketches show the public contract shapes. Validation and implementation details are omitted.

Definitions

Definitions describe reusable graph topology before concrete variable values are supplied. Materialization binds those values and produces an executable graph configuration.

class NodeDefinition(BaseModel):
    node_id: str
    node_type: str
    fields: tuple[NodeFieldSpec, ...]
    input_sources: dict[str, NodeInputSourceRef]
    output_field: str
    variable_names: frozenset[str]


class GraphDefinition(BaseModel):
    schema_version: Literal[1] = 1
    nodes: tuple[NodeDefinition, ...]
    terminal_node_id: str

def materialize(
    self,
    variable_assignments: Mapping[str, Mapping[str, Any]] | None = None,
) -> GraphConfig: ...

Configuration

Configurations are complete, validated graphs with concrete values. Their dependency structure has a deterministic topological order.

class NodeConfig(BaseModel):
    node_id: str
    node_type: str
    fields: tuple[NodeFieldSpec, ...]
    input_sources: dict[str, NodeInputSourceRef]
    output_field: str
    variables: dict[str, Any]


class GraphConfig(BaseModel):
    nodes: tuple[NodeConfig, ...]
    terminal_node_id: str

    def topological_order(self) -> tuple[NodeConfig, ...]: ...

def validate_graph_external_inputs(
    graph: GraphConfig,
    *,
    allowed_fields: Collection[str],
) -> None: ...

Identity

dr-graph owns graph definition, graph configuration, and graph-config identity. Every static configuration field participates in a versioned canonical identity document. dr-serialize turns that document into the graph's full SHA-256 hash. Callers may embed graph_hash as the inner or root layer of a larger canonical key. Scheduling, recovery, membership, and durable orchestration stay outside dr-graph.

GRAPH_CONFIG_IDENTITY_SCHEMA = "dr_graph.graph_config"
GRAPH_CONFIG_IDENTITY_SCHEMA_VERSION = 1


def graph_config_identity_document(
    graph: GraphConfig,
) -> IdentityDocument: ...


def graph_hash(graph: GraphConfig) -> str: ...

Execution

execute_graph is the sole graph interpretation entry point. It owns graph traversal and dependency wiring while the caller owns node behavior through RunNode, the only behavior delegation seam. New graph shapes need no new executor code. Intra-graph execution is serial; cross-run concurrency is caller- or platform-owned. A dependency-closed set of completed node outputs may be supplied to continue execution.

type RunNode = Callable[
    [NodeConfig, Mapping[str, Any]],
    NodeOutput | Mapping[str, Any],
]

def execute_graph(
    *,
    graph: GraphConfig,
    inputs: Mapping[str, Any],
    run_node: RunNode,
    completed: Mapping[str, NodeOutput | Mapping[str, Any]] | None = None,
) -> GraphRunResult: ...

Results

Results distinguish node outcomes from the aggregate graph outcome and retain enough structured state to inspect or continue a run.

class NodeOutcomeStatus(StrEnum):
    SUCCESS = "success"
    ERROR = "error"
    BLOCKED = "blocked"
    CANCELLED = "cancelled"


class NodeOutcomeSource(StrEnum):
    FRESH = "fresh"
    REUSED = "reused"


class GraphRunStatus(StrEnum):
    SUCCESS = "success"
    ERROR = "error"
    BLOCKED = "blocked"
    CANCELLED = "cancelled"

class NodeOutput(BaseModel):
    values: dict[str, Jsonable]
    metadata: dict[str, Jsonable]


class NodeOutcome(BaseModel):
    node_id: str
    status: NodeOutcomeStatus
    output: NodeOutput | None
    error: NodeError | None
    blocked_by: tuple[str, ...]
    outcome_source: NodeOutcomeSource

class GraphRunResult(BaseModel):
    graph_hash: str
    external_inputs: dict[str, Jsonable]
    status: GraphRunStatus
    outcomes: dict[str, NodeOutcome]
    execution_order: tuple[str, ...]
    terminal_node_id: str
    terminal_output: Jsonable
    terminal_error: TerminalError | None

Successful nodes invoked in the current run carry outcome_source=fresh. Completed outputs supplied through execute_graph(..., completed=...) are recorded as successful outcomes with outcome_source=reused.

Failure diagnostics

Node-behavior failures become NodeOutcome errors rather than escaping the graph run. dr-graph captures a strict-JSON NodeError snapshot from each exception. Per-leg success evidence flows through NodeOutput.metadata; fuller attempt evidence stays in caller-owned systems.

class NodeError(BaseModel):
    error_type: str
    message: str
    failure_class: str | None
    metadata: dict[str, Jsonable]
    traceback: str

error_type is always the real exception type (module.qualname). A caller-declared label may still be attached on the raising exception as an error_type attribute; when present, it is persisted in metadata["declared_error_type"] rather than replacing NodeError.error_type.

failure_class values are owned by the raising layer. dr-graph infrastructure errors such as InputResolutionError and NodeExecutionError carry failure_class="infrastructure". Callback exceptions may supply their own failure_class instance or class attribute.

NodeError.metadata retains every strict-JSON metadata entry independently. When a metadata key or value cannot be persisted, dr-graph records the loss in metadata["dropped_metadata"] as a list of {key, reason} entries rather than silently narrowing evidence. Reason literals include non_string_key, strict_json, and metadata_accessor_failed.

NodeOutput.metadata is the per-leg strict-JSON telemetry channel for successful node outputs. dr-graph validates it strictly and does not coerce or drop values.

Interruptions during execution (asyncio.CancelledError, KeyboardInterrupt) mark the in-flight node cancelled, block remaining nodes, build a partial GraphRunResult, and re-raise GraphRunInterruptedError with that result available as .partial_result before the interruption propagates.

class GraphRunInterruptedError(GraphExecutionError):
    partial_result: GraphRunResult

NodeOutcome.outcome_source records whether each successful outcome came from this run (fresh) or from caller-supplied completed outputs (reused).

Flow optimization

Flow optimization is independent of graph configuration and interpretation. The base package models declared network order and returns exact arc flows in that order.

class FlowArc:
    arc_id: ArcId
    source: NodeId
    target: NodeId
    capacity: int
    unit_cost: int


class FlowProblem:
    nodes: tuple[NodeId, ...]
    arcs: tuple[FlowArc, ...]
    source: NodeId
    sink: NodeId
    required_flow: int


class FlowResult:
    sent_flow: int
    total_cost: int
    arc_flows: tuple[ArcFlow, ...]

def solve_min_cost_flow(problem: FlowProblem) -> FlowResult: ...

The nested transportation package models each available route by its ordered marginal costs; every entry supplies one unit of capacity. Its result preserves source and destination index order as an allocation matrix.

class TransportCell:
    source_index: int
    destination_index: int
    marginal_costs: tuple[int, ...]


class TransportProblem:
    supplies: tuple[int, ...]
    demands: tuple[int, ...]
    cells: tuple[TransportCell, ...]


class TransportSolution:
    allocations: tuple[tuple[int, ...], ...]
    total_flow: int
    total_cost: int

def solve_separable_transport(
    problem: TransportProblem,
) -> TransportSolution: ...