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 gives every complete graph configuration a stable, versioned identity.
• Execution interprets a graph in topological order while delegating node behavior to the caller.
• 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

Every static configuration field participates in a versioned canonical identity document. dr-serialize turns that document into the graph's full SHA-256 hash.

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

Execution owns graph traversal and dependency wiring while the caller owns node behavior. 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"


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

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, ...]

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
    attempt_evidence_refs: tuple[str, ...]
    provenance: dict[str, Jsonable]

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: ...