fabric-graph

Fabric Graph

A graph is program code, not a framework. Pick the smallest form that fits:

• In-session graph: one Python fabric_exec program. Nodes are calls; edges are if, loops, bounded asyncio.gather, and budgets. Use it when the whole graph finishes in this turn.
• Resumable graph: a durable actor that runs one node per mailbox message, checkpoints each node with mesh.put CAS, and wakes itself for the next node. Use it when a node waits on a person or a schedule, or the run must survive Main and resident-host restarts.

Nodes that are reviewed, reusable steps can be saved programs: await programs.run(ref=ref, input=data, requirePromoted=True) runs one nested with the caller's capabilities.

In-session graph

import asyncio
target = π.target
types, tests = await asyncio.gather(
    pi.bash(command="bun run typecheck", settle=True),
    pi.bash(command="bunx vitest run " + target, settle=True),
)
if not types["ok"] or not tests["ok"]:
    return {"status": "failed", "node": "checks", "types": types["ok"], "tests": tests["ok"]}
review = await agents.run(task="Review the change to " + target + " for correctness. Reply APPROVE or a list of blockers.", name="graph review")
text = review.get("text") or ""
return {"status": "success" if review.get("status") == "completed" and text.startswith("APPROVE") else "partial", "review": text}

Pass the test target as payloads.target. A failed node returns its evidence; do not rerun passing nodes.

Resumable graph

Seed the run once, then create a durable actor that owns it. The state record is the single source of truth: runs/<run>/graph holds {node, status, results, decisionId?} with its mesh version.

run = π.run
key = "runs/" + run + "/graph"
await mesh.put(key=key, value={"node": "fetch", "status": "ready", "results": {}}, ifVersion=0)
instructions = "\n".join([
    "You own the resumable graph " + key + ". Nodes in order: fetch, test, approve, release.",
    "On every message, run exactly one fabric_exec program that:",
    "1. reads " + key + " with mesh.get and stops if status is done or failed;",
    "2. runs the current node only, writes its result and the next node with mesh.put(ifVersion=<the version you read>), and stops on a CAS conflict;",
    "3. for approve, raises tools.call(ref='decisions.raise', args={'title': ..., 'input': 'confirm'}) once, stores decisionId in the record, and waits with decisions.wait(id=..., timeoutMs=60000); while still open, schedules mesh.publish(topic=..., kind='graph.tick', afterMs=600000, key=...) and stops;",
    "4. after a successful node, wakes itself with agents.tell to its own id from mesh.self().",
    "Every node must be idempotent: before acting, check the recorded result; an interrupted spawn or shell run is indeterminate, so verify its effect before repeating it.",
])
actor = await agents.create(name="graph-" + run, runner="pi", residency="durable", topics=["graph." + run], instructions=instructions)
await agents.tell(id=actor["id"], message="Start graph " + key + ".")
return {"run": run, "key": key, "actor": actor["id"], "topic": "graph." + run}