Typed Handles & Reactive Engine

dagron ships seven coordinated additions that move beyond stringly-typed node addressing and add four headline differentiators no other Python DAG library combines: typed handles, a Tawazi-style flow API, fine-grained reactive recomputation, content-addressed cross-process caching, and time-travel replay. Existing string-based code keeps working — every new feature is opt-in.

1. NodeRef — typed node handles

dag.add_node() returns a stable NodeRef. Every public method that takes a node identifier accepts both str and NodeRef, so existing code keeps working.

from dagron import DAG, NodeRef

dag = DAG()
extract = dag.add_node("extract")     # NodeRef
transform = dag.add_node("transform") # NodeRef

dag.add_edge(extract, transform)      # NodeRef → NodeRef
dag.add_edge("extract", transform)    # str → NodeRef
dag.add_edge(extract, "transform")    # NodeRef → str

isinstance(extract, NodeRef)          # True
extract.name                          # "extract"
extract.epoch                         # 0

NodeRefs survive unrelated mutations (adding other nodes / edges) and detect remove-then-readd: removing "extract" and re-adding a node with the same name produces a NodeRef with a different epoch, so the old reference correctly raises StaleNodeRefError.

import pytest
from dagron import StaleNodeRefError

dag.remove_node(extract)
new_extract = dag.add_node("extract")  # fresh epoch
with pytest.raises(StaleNodeRefError):
    dag.has_edge(extract, transform)   # the old extract is stale

2. @dagron.flow — Pythonic compose API

Build a DAG by writing a regular Python function. Each @task call inside a @flow body records a node; passing one task's return value to another wires the edge. No string IDs, no fluent builder — just Python.

import dagron

@dagron.task
def fetch() -> list[int]:
    return [1, 2, 3, 4]

@dagron.task
def total(rows: list[int]) -> int:
    return sum(rows)

@dagron.task
def label(value: int) -> str:
    return f"Total = {value}"

@dagron.flow
def pipeline():
    return label(total(fetch()))

dag = pipeline.dag()                 # the underlying DAG, for analysis
result = pipeline()                  # builds + runs → ExecutionResult
result["label"].result               # "Total = 10"

The same @task decorator is compatible with the legacy parameter-name inference of Pipeline, so a single set of tasks can power both APIs. Inside a @flow context, calling transform(raw) returns a FlowFuture[T] placeholder; outside one, it executes normally.

3. Generic typing & dagron.stubgen

FlowFuture[T] and NodeResult[T] carry the wrapped task's return type all the way through:

from dagron import FlowFuture
from dagron.execution._types import NodeResult

@dagron.task
def fetch() -> list[int]: ...

@dagron.task
def total(rows: list[int]) -> int: ...

@dagron.flow
def pipeline():
    raw = fetch()                 # type-checks as list[int]
    return total(raw)             # type-checks as int

result = pipeline()
result[fetch].result              # NodeResult[list[int]] → list[int]
result[total].result              # NodeResult[int]       → int

For string-keyed lookups, generate a stub:

from dagron.stubgen import generate_stub

stub = generate_stub(
    pipeline.dag(),
    tasks={"fetch": fetch, "total": total},
    name="PipelineResult",
)
print(stub)
# class PipelineResult:
#     @overload
#     def __getitem__(self, key: Literal['fetch']) -> NodeResult[list[int]]: ...
#     @overload
#     def __getitem__(self, key: Literal['total']) -> NodeResult[int]: ...

Save the output as a .pyi file alongside your code; mypy will type result["fetch"] as NodeResult[list[int]] even though result itself is just ExecutionResult.

4. Effect tags

Tag every @task with its side-effect class — the engine uses these for parallelism gating today and for cache / replay semantics in the features below.

from dagron import Effect

@dagron.task                                     # defaults to Effect.PURE
def add(a: int, b: int) -> int: return a + b

@dagron.task(effect=Effect.NETWORK)
def fetch_user(uid: int) -> dict: ...

@dagron.task(effect=Effect.NONDETERMINISTIC)
def now() -> float:
    import time; return time.time()

Properties:

@flow mirrors each task's effect onto its DAG node's metadata; read back with dagron.effects_of(dag). An AST-scan heuristic emits a UserWarning when a PURE task appears to call impure functions (time.time, random.*, os.*, requests.*, …).

DAGExecutor(enforce_effect_isolation=True) serializes NONDETERMINISTIC tasks while letting other effects parallelize freely.

5. Reactive engine — Signal / Computed / Watcher

dagron.reactive provides Solid.js / Jane-Street-Incremental style primitives where the dependency graph is implicit: building a Computed records its read dependencies as a side-effect of evaluating the function.

import dagron.reactive as dr

a = dr.signal(1)
b = dr.signal(2)
s = dr.computed(lambda: a() + b())
p = dr.computed(lambda: s() * 10)

p()                  # 30 — initial compute
a.set(5)             # invalidates s and p; b untouched
p()                  # 70 — recomputes only s and p

@dr.watch
def watch_p():
    print("p =", p())

with dr.batch():     # glitch-free
    a.set(0)
    b.set(0)
# watch_p fires exactly once after the batch, sees p == 0

Headline benchmark: in a graph of 10,000 derived nodes off one root signal, mutating the root and reading just one branch takes ~10 µs — the engine recomputes only the read path, skipping the other 9999 invalidated-but-unread branches. This is the differentiator no other Python DAG library delivers.

This module is distinct from the existing dagron.execution.reactive.ReactiveDAG, which wraps a pre-built dagron.DAG and exposes a push-based subscribe() / set_input() API. Use whichever fits your shape: the reactive primitives for fresh dependency graphs you build in code; ReactiveDAG to layer reactivity over a DAG you already have.

6. Cross-process content-addressed cache

dagron.contentcache is Nix-flake-style: the cache is keyed by content hash, the filesystem path is the index, and there's no index.json to keep in sync. Independent processes share intermediates transparently — a build on one CI worker hits the cache on another the moment they compute the same fingerprint.

from dagron import Effect
from dagron.contentcache import ContentCache

cache = ContentCache()       # ~/.cache/dagron/cas

def expensive(x: int) -> int:
    return x * 1000

# First call: miss, computes, writes payload to CAS.
val, hit = cache.compute_or_cached(expensive, args=(42,), effect=Effect.PURE)
# In another process / another day:
val, hit = cache.compute_or_cached(expensive, args=(42,), effect=Effect.PURE)
# `hit` is True; the payload deserialized straight from disk.

Effect-aware: WRITE / NETWORK / NONDETERMINISTIC tasks bypass the cache entirely (their results aren't reproducible). Pluggable via the Hasher protocol — default_hash (pickle + blake2b) handles most Python types; numpy_hash uses array.tobytes() for byte equality; write your own for polars frames or any tobyte-friendly type. Honors $DAGRON_CACHE_DIR.

7. Time-travel replay

dagron.trace writes an append-only JSONL log of node executions; each record references a payload stored by fingerprint in the ContentCache, so identical values across runs deduplicate automatically. replay(at=t) walks the log up to time t and reconstructs the per-node state.

from dagron.contentcache import ContentCache
from dagron.trace import TraceWriter, replay

cas = ContentCache()
log_path = "run-2026-05.jsonl"

with TraceWriter(log_path, cas=cas) as w:
    w.record("fetch",     value=[1, 2, 3], effect=Effect.PURE,    timestamp=t0)
    w.record("transform", value=6,         effect=Effect.PURE,    timestamp=t0 + 1)
    w.record("publish",   value="ok",      effect=Effect.NETWORK, timestamp=t0 + 2)

# Days later, in another process:
state = replay(log_path, at=t0 + 1.5, cas=cas)
state["fetch"].value         # [1, 2, 3] — byte-identical to the original run
state["transform"].value     # 6
"publish" in state           # False — cutoff was before publish ran

state = replay(log_path, cas=cas)
state["publish"].value         # "ok" — surfaced from the log
state["publish"].replayable    # False — NETWORK is non-deterministic

Pure / READ nodes replay byte-identically. Impure nodes (WRITE/NETWORK/NONDETERMINISTIC) are flagged replayable=False but their logged values are still exposed, so you can audit what the run actually produced. Honors $DAGRON_TRACE_DIR.

How they fit together

The seven additions are designed to compose:

• NodeRef is the substrate — every later API references nodes by the typed handle.
• @flow records call structure into a dagron.DAG, mirroring each task's effect onto node metadata.
• stubgen turns the @flow-built DAG into a typed lookup stub.
• Effects drive parallelism isolation, cache opt-in, and replay reproducibility flags — one tag, three downstream behaviours.
• Reactive is the "live" face of computation; content cache is its persistent face; replay is its retrospective face.

You can adopt any subset independently. The string-based DAG API, Pipeline, and the existing ReactiveDAG / ContentAddressableCache classes remain unchanged.