graph

type AppendReducer ¶ added in v0.3.0

type AppendReducer struct{}

AppendReducer appends update elements to an existing slice. If the existing value is not a slice, it is replaced. If the update is a single value, it is wrapped in a slice and appended.

func (AppendReducer) Reduce(existing, update any) any

type CheckpointCommitter ¶ added in v0.4.0

type CheckpointCommitter interface {
	SaveCheckpointAndEvent(ctx context.Context, cp *storage.Checkpoint, evt *storage.Event) error
}

CheckpointCommitter is an optional interface a storage.Storage implementation may satisfy to persist a checkpoint and its ledger event atomically in a single transaction. When the configured store implements it, the runner uses it so a crash between the two writes cannot desync the checkpoint from the event ledger. Stores that do not implement it fall back to two idempotent calls.

type CompiledGraph ¶

type CompiledGraph struct {
	ID      string
	Nodes   map[string]*Node
	AdjList map[string][]*Edge // from -> edges
	Entry   string
}

CompiledGraph is the immutable, validated graph ready for execution.

func RegisterEntrypoint(name string, fn EntrypointFunc) (*CompiledGraph, error)

func TaskGraph(id string, tasks map[string]TaskFunc) (*CompiledGraph, error)

func (cg *CompiledGraph) ToDOT() string

func (cg *CompiledGraph) ToMermaid() string

type Edge ¶

type Edge struct {
	From      string
	To        string        // static target (mutually exclusive with Condition)
	Condition EdgeCondition // dynamic routing (mutually exclusive with To)
}

Edge represents a transition between nodes.

type EdgeCondition ¶

type EdgeCondition func(state State) string

EdgeCondition decides which node to transition to based on state. Returns the target node ID.

type EntrypointFunc ¶ added in v0.3.0

type EntrypointFunc func(ctx context.Context, input any) (any, error)

EntrypointFunc is a function that serves as a graph entrypoint.

type GraphResolver ¶ added in v0.4.0

type GraphResolver func(graphID string) (*CompiledGraph, error)

GraphResolver returns the compiled graph for a graph ID. A distributed worker uses it to look up the graph definition it must execute for a claimed run.

func SingleGraphResolver(g *CompiledGraph) GraphResolver

type MergeFunc ¶ added in v0.3.0

type MergeFunc func(original State, results []State) (State, error)

MergeFunc combines the results of parallel branches into a single state. It receives the original pre-fork state and the results from each branch.

type MergeMapReducer ¶ added in v0.3.0

type MergeMapReducer struct{}

MergeMapReducer performs a shallow merge of two maps. Update keys overwrite existing keys. Non-map values are replaced.

func (MergeMapReducer) Reduce(existing, update any) any

type Message ¶

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

Message records a message exchanged during the run.

type Node ¶

type Node struct {
	ID string
	Fn NodeFunc
	// Interrupt, if true, causes the runner to checkpoint and pause before executing this node.
	Interrupt bool
}

Node represents a node in the state graph.

type NodeFunc ¶

type NodeFunc func(ctx context.Context, state State) (State, error)

NodeFunc is the function signature for a graph node. It receives the current state and returns updated state.

func FanOut(branches []NodeFunc, merge MergeFunc) NodeFunc

func RegisterTask(name string, fn TaskFunc) NodeFunc

func SubgraphNode(sub *CompiledGraph) NodeFunc

type QueuedExecutor ¶ added in v0.4.0

type QueuedExecutor struct {
	// contains filtered or unexported fields
}

QueuedExecutor bridges the durable work queue to the existing graph Runner. It implements the intake/execution split: producers enqueue runs and this executor — invoked by any queue.Worker — runs or resumes the graph via a freshly constructed Runner, exactly as a synchronous caller would. The public Runner API is unchanged; this only calls it.

func NewQueuedExecutor(store storage.Storage, resolve GraphResolver) *QueuedExecutor

func (qe *QueuedExecutor) Executor() queue.Executor

func (qe *QueuedExecutor) WithMaxSteps(n int) *QueuedExecutor

type Reducer ¶ added in v0.3.0

type Reducer interface {
	Reduce(existing, update any) any
}

Reducer defines how a state value is combined when a node returns an update. The default behavior (nil reducer) is full replacement, but specific keys can be configured to append to slices, merge maps, etc.

type ReducerFunc ¶ added in v0.3.0

type ReducerFunc func(existing, update any) any

ReducerFunc is a convenience adapter for functions.

func (f ReducerFunc) Reduce(existing, update any) any

type ReplaceReducer ¶ added in v0.3.0

type ReplaceReducer struct{}

ReplaceReducer always replaces the existing value with the update. This is the default behavior when no reducer is configured.

func (ReplaceReducer) Reduce(_, update any) any

type RunPayload ¶ added in v0.4.0

type RunPayload struct {
	Initial State `json:"initial,omitempty"`
	Resume  bool  `json:"resume,omitempty"`
}

RunPayload is the durable envelope carried by a queued graph run. It is stored as the queue.Run payload so a worker on any node can reconstruct the execution without shared in-process state. When Resume is true the executor continues from the session's latest checkpoint instead of starting fresh; this is how a human-in-the-loop run, once parked and later signaled, advances past its interrupt rather than restarting.

type RunState ¶

type RunState struct {
	RunID       string    `json:"run_id"`
	SessionID   string    `json:"session_id"`
	GraphID     string    `json:"graph_id"`
	CurrentNode string    `json:"current_node"`
	Status      RunStatus `json:"status"`
	State       State     `json:"state"`
	SeqNum      int64     `json:"seq_num"`
	StartedAt   time.Time `json:"started_at"`
	UpdatedAt   time.Time `json:"updated_at"`

	// Rich output (Agno-inspired)
	Messages   []Message        `json:"messages,omitempty"`
	ToolCalls  []ToolCallRecord `json:"tool_calls,omitempty"`
	TotalUsage UsageStats       `json:"total_usage"`
}

RunState captures the full runtime state for a single graph execution.

type RunStatus ¶

type RunStatus string

RunStatus represents the status of a graph run.

const (
	RunStatusRunning   RunStatus = "running"
	RunStatusPaused    RunStatus = "paused"
	RunStatusCompleted RunStatus = "completed"
	RunStatusFailed    RunStatus = "failed"
)

type Runner ¶

type Runner struct {
	// contains filtered or unexported fields
}

Runner executes a CompiledGraph with durable checkpointing.

A Runner is single-use: each Run/Resume/Replay/Fork call must be made on a freshly constructed Runner. Reusing a Runner returns an error rather than panicking on a closed channel.

func NewRunner(g *CompiledGraph, store storage.Storage) *Runner

func (r *Runner) ForkFrom(ctx context.Context, checkpointID string, stateUpdate map[string]any) (*RunState, error)

func (r *Runner) ReplayFrom(ctx context.Context, checkpointID string) (*RunState, error)

func (r *Runner) Resume(ctx context.Context, sessionID string) (*RunState, error)

func (r *Runner) ResumeFromCheckpoint(ctx context.Context, checkpointID string) (*RunState, error)

func (r *Runner) Run(ctx context.Context, sessionID string, initial State) (*RunState, error)

func (r *Runner) Stream() <-chan StreamEvent

func (r *Runner) WithBroker(b *stream.Broker) *Runner

func (r *Runner) WithMaxSteps(n int) *Runner

func (r *Runner) WithNodeTimeout(d time.Duration) *Runner

func (r *Runner) WithTracer(t Tracer) *Runner

type State ¶

type State map[string]any

State is the mutable state bag passed through graph execution.

func MergeAll(original State, results []State) (State, error)

type StateGraph ¶

type StateGraph struct {
	// contains filtered or unexported fields
}

StateGraph is a builder for defining directed graphs with durable execution.

func New(id string) *StateGraph

func (g *StateGraph) AddConditionalEdge(from string, condition EdgeCondition) *StateGraph

func (g *StateGraph) AddEdge(from, to string) *StateGraph

func (g *StateGraph) AddInterruptNode(id string, fn NodeFunc) *StateGraph

func (g *StateGraph) AddNode(id string, fn NodeFunc) *StateGraph

func (g *StateGraph) AddSubgraph(id string, sub *CompiledGraph) *StateGraph

func (g *StateGraph) Compile() (*CompiledGraph, error)

func (g *StateGraph) SetEntryPoint(nodeID string) *StateGraph

func (g *StateGraph) SetFinishPoint(nodeID string) *StateGraph

type StateSchema ¶ added in v0.3.0

type StateSchema map[string]Reducer

StateSchema maps state keys to their reducers. Keys not in the schema use the default ReplaceReducer behavior.

func (s StateSchema) ApplyUpdate(existing, update State) State

type StreamEvent ¶

type StreamEvent struct {
	Type      string    `json:"type"` // node_start, node_end, edge_transition, checkpoint, interrupt, error
	NodeID    string    `json:"node_id,omitempty"`
	State     State     `json:"state,omitempty"`
	Error     string    `json:"error,omitempty"`
	Timestamp time.Time `json:"timestamp"`
}

StreamEvent is emitted during graph execution for real-time observability.

type TaskFunc ¶ added in v0.3.0

type TaskFunc func(ctx context.Context, input any) (any, error)

TaskFunc is a function that serves as a checkpoint-able task.

type ToolCallRecord ¶

type ToolCallRecord struct {
	Name   string         `json:"name"`
	Args   map[string]any `json:"args"`
	Result any            `json:"result"`
	Error  string         `json:"error,omitempty"`
}

ToolCallRecord records a tool invocation during the run.

type Tracer ¶ added in v0.4.0

type Tracer interface {
	StartSpan(ctx context.Context, sessionID, name, kind string) (*storage.Trace, error)
	EndSpan(ctx context.Context, t *storage.Trace, output any, errMsg string) error
}

Tracer records execution spans. It is defined here (rather than importing the os/trace control-plane package) so the engine layer never depends upward on os/; the ChronosOS *trace.Collector satisfies it. Both methods use storage.Trace, which the engine already depends on.

type UsageStats ¶

type UsageStats struct {
	PromptTokens     int `json:"prompt_tokens"`
	CompletionTokens int `json:"completion_tokens"`
	TotalTokens      int `json:"total_tokens"`
}

UsageStats tracks cumulative token usage across a run.