01
Nothing lives only in memory
Every state transition is a compare-and-swap write to the executions KV bucket. Crash an instance mid-flight and another resumes exactly where it left off.
Packtrail orchestrates declarative flow graphs — task, fanout,
fanin, choice and signal nodes, defined in
YAML or as Go structs — with full crash durability backed only by
NATS. Node execution runs through a pluggable Invoker: plug in your
own agent caller, an HTTP client, or the built-in NATS worker and inherit all of
packtrail's machinery for free.
Four commitments shape every line of the engine.
Every state transition is a compare-and-swap write to the executions KV bucket. Crash an instance mid-flight and another resumes exactly where it left off.
JetStream KV holds execution state, streams give work-queue durability, the Message Scheduler handles timers. No second datastore to run, scale or back up.
The engine never speaks a wire protocol directly. Every task node runs through an Invoker — plug in any transport and inherit all of packtrail's durability machinery.
Signals dedupe by JetStream sequence. The optional result cache memoises invocations by (execution, node, visit, attempt) so reruns never re-invoke a side effect.
Each task node dispatches to the configured Invoker, with per-node timeouts and exponential / linear / fixed retry backoff scheduled durably in NATS.
Run branches in parallel and join with all, any or quorum:N. Finished branches are persisted in KV and never recomputed on engine takeover.
choice nodes evaluate expr-lang expressions over the assembled context — input, results, signals — first match wins, expressions compiled at startup, not at runtime.
signal nodes pause an execution until an external event arrives — durable and idempotent, with an on_timeout fallback route when the deadline passes.
Slow nodes never hold a work slot. The built-in asyncqueue invoker makes any ordinary Invoker durable and asynchronous — dispatch to a work-queue, run on a worker pool, settle via CompleteActivity, at-least-once. Or return StatusPending yourself.
Recurring flow starts, retry backoff and signal timeouts all run through the JetStream Message Scheduler. The engine never keeps an in-process timer — they survive restarts.
The engine walks the flow graph and calls the configured Invoker for each node — it never
speaks a wire protocol itself. Register any number of Invoker kinds; each node selects
one via its invoker: field. All state lives in NATS.
packtrail-executions
KV · control plane, source of truth
packtrail-payloads
KV · data plane: inputs, outputs, signals
packtrail-leases
KV · ownership (TTL)
packtrail-work
stream · work queue
packtrail-events
stream · domain events
packtrail-signals
stream · external signals
packtrail-schedule
scheduler · timers & cron
packtrail-flows
KV · flow registry
packtrail-result-cache
KV · idempotency (opt-in)
packtrail-executions-archive
KV · cold archive (opt-in)
packtrail-deadletter
stream · dead letters (~30d)
packtrail-history
stream · per-exec trace (opt-in)
packtrail-idx-status
KV · projection
packtrail-idx-flow
KV · projection
A durable pull consumer distributes steps across every engine instance; the per-execution lease narrows concurrent processing and the execution-doc CAS fences state. Invocation stays at-least-once — enable the result cache for non-idempotent targets.
Each write is conditioned on the execution's KV revision. Concurrent fan-out branches retry their read-modify-write rather than block, so the hot path stays lock-free and crash-safe.
The engine has no dependency on any transport — enforced at the module level by a CI acceptance test. Any project plugs in its own Invoker and inherits everything else.
Each flow's graph is published to packtrail-flows at startup, so observability tools like packtrail-ui can render the live graph without access to the source files or the engine process.
Implement a single method and your services — AI agents, HTTP endpoints, message queues — gain crash durability, retries, fan-in policies, signals and timers at no extra cost.
// Invoker is the single seam between packtrail and your ecosystem.
type Invoker interface {
Invoke(ctx context.Context, req packtrail.Request) (packtrail.Result, error)
}
// Request carries everything the Invoker needs:
// req.Target → agent name, URL, subject …
// req.ExecutionID → owning execution
// req.Payload → assembled {input, results, signals} context
// req.Attempt → 0-based retry counter
// req.Deadline → per-node timeout
// --- Example: custom agent invoker ---
type agentInvoker struct{ client *AgentClient }
func (a *agentInvoker) Invoke(
ctx context.Context, req packtrail.Request,
) (packtrail.Result, error) {
out, err := a.client.Call(ctx,
req.Target, req.Payload)
if err != nil {
return packtrail.Result{}, err // → retry per node policy
}
return packtrail.Result{
Status: packtrail.StatusOK,
Payload: out,
}, nil
}
// Register it once at startup:
packtrail.WithInvoker("agent", &agentInvoker{client: c})// WithResultCache() wraps the registry in a KV-backed cache keyed by
// (execution, node, visit generation, attempt). A redelivery after a
// crash returns the stored result instead of re-invoking. A genuine
// retry (new attempt) or a re-visit via a legal cycle / Resume (new
// generation) still calls the Invoker. Enable whenever invocations
// have side effects.
srv, _ := packtrail.New(nc,
packtrail.WithFlowsDir("flows"),
packtrail.WithInvoker("agent", &agentInvoker{client: c}),
packtrail.WithResultCache(), // ← one option, full idempotency
)Define flows in YAML or as Go structs — both paths run through the same validation and produce identical runtime behaviour.
version: "1.0"
name: agent-pipeline
nodes:
- {id: triage, type: task,
invoker: agent, target: triage-agent,
timeout: 2m,
retry: {max_attempts: 3, backoff: exponential}}
- {id: research, type: fanout,
branches: [tech, market, legal]}
- {id: tech, type: task, invoker: agent, target: tech-agent}
- {id: market, type: task, invoker: agent, target: market-agent}
- {id: legal, type: task, invoker: agent, target: legal-agent}
- {id: join, type: fanin,
wait_for: [tech, market, legal], join_policy: all}
- id: route
type: choice
rules:
- {when: "results.triage.risk_score > 80", to: escalation}
- {default: true, to: synthesis}
- {id: synthesis, type: task, invoker: agent, target: synthesis-agent}
# Built-in nats-task: subject instead of invoker+target
- {id: escalation, type: task,
subject: "tasks.escalate.{execution_id}"}
- {id: gate, type: signal, signal_name: approval,
timeout: 24h, on_timeout: escalation}
edges:
- {from: triage, to: research}
- {from: research, to: join}
- {from: join, to: route}
- {from: synthesis, to: gate}packtrail.WithFlowDef(packtrail.FlowDef{
Name: "agent-pipeline",
Nodes: []packtrail.NodeDef{
{ID: "triage", Type: "task", Invoker: "agent", Target: "triage-agent",
Timeout: 2*time.Minute,
Retry: &packtrail.RetryPolicy{MaxAttempts: 3, Backoff: "exponential"}},
{ID: "research", Type: "fanout", Branches: []string{"tech", "market", "legal"}},
{ID: "tech", Type: "task", Invoker: "agent", Target: "tech-agent"},
{ID: "market", Type: "task", Invoker: "agent", Target: "market-agent"},
{ID: "legal", Type: "task", Invoker: "agent", Target: "legal-agent"},
{ID: "join", Type: "fanin",
WaitFor: []string{"tech", "market", "legal"}, JoinPolicy: "all"},
{ID: "route", Type: "choice", Rules: []packtrail.RuleDef{
{When: "results.triage.risk_score > 80", To: "escalation"},
{Default: true, To: "synthesis"},
}},
{ID: "synthesis", Type: "task", Invoker: "agent", Target: "synthesis-agent"},
{ID: "escalation", Type: "task", Subject: "tasks.escalate.{execution_id}"},
{ID: "gate", Type: "signal", SignalName: "approval",
Timeout: 24*time.Hour, OnTimeout: "escalation"},
},
Edges: []packtrail.EdgeDef{
{From: "triage", To: "research"},
{From: "research", To: "join"},
{From: "join", To: "route"},
{From: "synthesis", To: "gate"},
},
})
An Invoker that returns StatusPending parks the execution without blocking
a work slot; the worker calls CompleteActivity when done —
idempotent, stale-safe, for task nodes and fan-out branches alike. The built-in
invoker/asyncqueue package does all of this for you — the manual
mechanism is shown here.
// dispatch (non-blocking): enqueue durable work, return pending.
func (d *dispatcher) Invoke(
ctx context.Context, req packtrail.Request,
) (packtrail.Result, error) {
// Enqueue into your durable queue (DB, NATS, SQS …)
enqueueJob(req.ExecutionID, req.NodeID,
req.Attempt, req.Payload)
// Engine parks the execution (status=waiting) and
// frees its work slot immediately.
return packtrail.Result{Status: packtrail.StatusPending}, nil
}
// later, from the worker that finished the job:
srv.CompleteActivity(ctx,
execID, nodeID, attempt,
packtrail.Result{
Status: packtrail.StatusOK,
Payload: out,
})
// CompleteActivity is idempotent and stale-safe:
// a duplicate call or one that arrives for the wrong
// attempt is a silent no-op — safe for at-least-once workers.You rarely write the code on the left. WithAsyncInvoker("kind", exec) takes an ordinary synchronous Invoker and makes it durable and async — a JetStream work-queue, a hosted worker pool, dispatch dedup, ack-extending heartbeats and crash redelivery, all handled.
The engine parks the execution as waiting and moves on to other work immediately. No goroutine is held, no work slot consumed — a single engine instance can have thousands of inflight async activities.
CompleteActivity is keyed by (execution, node, attempt) — and CompleteActivityWithGeneration also fences the node-visit generation, for flows with legal cycles or Resume. A duplicate or stale completion from an at-least-once worker is a silent no-op. A completion that races ahead of the parking step is stashed and consumed on arrival.
Each fan-out branch runs through the Invoker independently. Return StatusPending from any or all branches — they park individually, and the fan-in join re-evaluates whenever a branch settles.
The packtrail package is the public entry point. Co-locate the engine and your
task workers in a single binary — still crash-durable because every transition is a
write to NATS, still scalable to N replicas.
packtrail-ui is a read-only web dashboard that connects to the same NATS cluster,
reads execution state from the flow registry KV, and tails the live event stream —
no engine process or source files required.
Browse all executions, filter by status or flow, drill into any running or failed instance.
The complete flow graph is rendered as SVG with the current execution position highlighted in real time over SSE.
Status, current node, stored outputs, branch states, received signals, and the error message on failure — all in one place. A dead-letter tile surfaces poisoned work.
Server-Sent Events push every execution transition to the browser — no polling, no page reload needed.
# NATS_URL env is honoured (defaults to localhost:4222)
$ go run ./cmd/packtrail-ui \
--namespace packtrail \
--addr :8088| Endpoint | Returns |
|---|---|
| GET /api/flows | Names of all registered flows |
| GET /api/flows/{name} | Full flow graph (nodes + edges) as FlowGraph |
| GET /api/executions | Execution summaries; filter with ?status= or ?flow= |
| GET /api/executions/{id} | Execution control-state snapshot including branches and signals |
| GET /api/executions/{id}/results | Assembled {input, results, signals} context (data plane) |
| GET /api/executions/{id}/history | Ordered transition trace (empty unless WithHistory) |
| GET /api/deadletters | Dead-letter count and recent dead-letter records |
| GET /api/events | Live Server-Sent Events stream of every execution transition |
The same data is available via the Server API: ListFlows, FlowGraph, Get / ByStatus / ByFlow, WatchEvents — a live channel of transitions — plus History, DeadLetterCount and RecentDeadLetters.
Because each flow's graph is published to packtrail-flows KV at engine startup, packtrail-ui can render it without access to the YAML files or the running engine process.