workflow

package
v0.1.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 18, 2026 License: Apache-2.0 Imports: 23 Imported by: 0

Documentation

Overview

Package workflow compiles versioned, declarative workflow definitions into immutable in-process execution plans.

Workflow is independent of the ai and agent packages. Applications provide behavior through explicitly registered Actions; definitions contain data, bindings, and control edges, never executable code.

Built-in composition remains fixed-flow and synchronous: Selector chooses one ordered route, SubWorkflow executes an exact resolved revision, Batch maps an inline child Definition over an array with bounded workers, and Loop sequentially executes a bounded inline body with transactional local variables. Batch and Loop are distinct and cannot contain each other. Merge v1 preserves exactly-one-present branch aggregation; Merge v2 uses ordered first-non-null selection for route-relevant upstream candidates. Both versions keep parallel Merge as direct wait-all fan-in.

Runs can stop at compile-time before/after boundaries, at dynamic Interrupt calls, or on a host signal. A CheckpointStore persists one opaque snapshot per Run ID so a new Runner can validate the same Plan and Resume the exact root and composite frontier. Checkpoints are process-runtime state, not part of the Definition wire contract.

During a real node invocation, Actions and custom NodeTypes can call GetExecutionContext to read the current Run, Definition, scoped node address, and cumulative one-based attempt. The returned address is detached runtime correlation metadata; hosts remain responsible for defining idempotency at their own side-effect boundaries.

Exhausted node failures either stop execution, select a dedicated error route, or continue with compile-validated default outputs. Error-route consumers use BindingNodeError to read the bounded error_message and error_type ports; normal outputs remain unavailable on that path. Handled nodes have NodeStatusException, and a completed root execution that observed one has RunStatusPartialSucceeded. Error messages are sensitive Workflow data and never lifecycle Event payloads.

RunPartial executes the root dependency slice ending at one destination. Hosts may supply detached successful data from a prior Partial Run, current pinned outputs, and directly changed node IDs; the runtime validates routes and schemas, invalidates dirty descendants, and executes every remaining boundary through the ordinary scheduler. Composite nodes remain atomic. ResumePartial uses the same opaque CheckpointStore protocol, while Run and Resume never consult Partial Run data.

PrepareNodeDebug derives a Coze-style isolated trial plan for one eligible node. DebugNode executes that selected node without executing its upstream graph; outside bindings become explicit final-schema inputs while literals and selected composite child Plans remain internal. Results contain sensitive resolved values and Action error text, so hosts own authorization, redaction, encryption, and retention. ResumeNodeDebug uses the same opaque CheckpointStore boundary as Resume. Node Debug invokes real Actions and is distinct from partial graph replay or pinned-data execution.

SchemaV1Alpha1 preserves required schema-only Workflow inputs. V1Alpha2 adds required, optional, and non-null default input contracts while retaining strict version-directed decoding. Defaults normalize only at a Workflow boundary, never as generic Action or NodeType port behavior. Events are process-local observations and intentionally have no JSON wire format.

Compile reports Definition diagnostics as a *CompileError containing stable issue codes and Definition, node, or control-path locations. Diagnostics are phase-gated: Compile returns all safely independent issues in the earliest failed compiler phase, not every hypothetical downstream problem. Nested SubWorkflow, Batch, and Loop issues retain full NodePath locations. Issues are process-local values with no prescribed JSON or persistence format; hosts own localization and versioned API, database, or editor projections.

WithNodeExecutionRecorder exposes detached running and effective outcome snapshots for ordinary Run and Resume operations. The callback is a synchronous, process-local upsert boundary keyed by RunID and NodeAddress; the host owns persistence failures, timeouts, redaction, retention, and any stable transport schema. Node Debug and Partial Run do not call it.

Index

Examples

Constants

View Source 
const (
	// SchemaV1Alpha1 identifies the first Workflow Definition wire format.
	SchemaV1Alpha1 = "pips.workflow/v1alpha1"
	// SchemaV1Alpha2 adds required, optional, and default Workflow inputs.
	SchemaV1Alpha2 = "pips.workflow/v1alpha2"
)
View Source 
const (
	// NodeErrorMessagePort contains the terminal error text after retry exhaustion.
	NodeErrorMessagePort = "error_message"
	// NodeErrorTypePort contains the stable [FailureKind] classification.
	NodeErrorTypePort = "error_type"
)

Variables

View Source 
var (
	// ErrInvalidValue classifies malformed or unsupported JSON values.
	ErrInvalidValue = errors.New("workflow: invalid value")
	// ErrInvalidSchema classifies malformed or unsupported port schemas.
	ErrInvalidSchema = errors.New("workflow: invalid schema")
	// ErrSchemaViolation means a Value does not satisfy a PortSchema.
	ErrSchemaViolation = errors.New("workflow: schema violation")
	// ErrInvalidDefinition classifies malformed workflow definitions.
	ErrInvalidDefinition = errors.New("workflow: invalid definition")
	// ErrInvalidRegistry classifies invalid node type or Action registrations.
	ErrInvalidRegistry = errors.New("workflow: invalid registry")
	// ErrCompile classifies a definition that cannot produce an execution plan.
	ErrCompile = errors.New("workflow: compile failed")
	// ErrRun classifies a workflow execution failure.
	ErrRun = errors.New("workflow: run failed")
	// ErrInterrupted classifies a successfully checkpointed, resumable Run.
	ErrInterrupted = errors.New("workflow: interrupted")
	// ErrEventWireFormat reports an attempt to serialize a process-local Event.
	ErrEventWireFormat = errors.New("workflow: event has no wire format")
	// ErrNodeExecutionWireFormat reports an attempt to serialize a sensitive,
	// process-local NodeExecution snapshot.
	ErrNodeExecutionWireFormat = errors.New("workflow: node execution has no wire format")
)

Workflow validation and execution errors. Match them with errors.Is.

Functions

func CompositeInterrupt 

func CompositeInterrupt(
	ctx context.Context,
	info Value,
	state Value,
	causes ...error,
) error

CompositeInterrupt propagates descendant interruption errors while saving local composite state. Ordinary errors are rejected.

func DecodeValue 

func DecodeValue[T any](v Value) (T, error)

DecodeValue decodes v into T using encoding/json rules.

func Interrupt 

func Interrupt(ctx context.Context, info Value) error

Interrupt suspends the current invocation with caller-facing information.

func StatefulInterrupt 

func StatefulInterrupt(ctx context.Context, info, state Value) error

StatefulInterrupt suspends the current invocation and persists private local state for GetInterruptState on a targeted resume.

func WithRunInterrupt 

func WithRunInterrupt(
	parent context.Context,
) (context.Context, func(...RunInterruptOption))

WithRunInterrupt enables a resumable host interruption signal without canceling the parent context. The returned function is safe to call more than once; only its first call takes effect.

Types

type Action 

type Action interface {
	Spec() ActionSpec
	Run(context.Context, ActionInput) (ActionOutput, error)
}

Action is trusted host behavior referenced by key and exact version from an ActionNode Definition. Implementations must be safe for concurrent Runs, honor context cancellation, and avoid retaining or mutating ActionInput.

type ActionConfig 

type ActionConfig struct {
	Action     ActionKey `json:"action"`
	Version    string    `json:"version"`
	Parameters *Value    `json:"parameters,omitempty"`
}

ActionConfig selects one exact Action and carries optional immutable parameters that are separate from data bindings.

type ActionInput 

type ActionInput struct {
	Values     map[string]Value
	Parameters Value
}

ActionInput carries resolved values and optional immutable node parameters.

type ActionKey 

type ActionKey string

ActionKey identifies a registered host Action family.

type ActionNode 

type ActionNode struct{}

ActionNode invokes one host-registered Action.

func (ActionNode) Compile 

func (ActionNode) Compile(
	_ context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (ActionNode) Spec 

func (ActionNode) Spec() NodeTypeSpec

Spec implements NodeType.

type ActionOutput 

type ActionOutput struct {
	Values map[string]Value
}

ActionOutput contains values produced by an Action.

type ActionSpec 

type ActionSpec struct {
	Key     ActionKey
	Version string
	Inputs  map[string]PortSchema
	Outputs map[string]PortSchema
}

ActionSpec identifies one host Action and its data contract.

type BatchConfig 

type BatchConfig struct {
	Body           Definition     `json:"body"`
	ResultOutput   string         `json:"result_output"`
	Mode           BatchMode      `json:"mode"`
	MaxConcurrency int            `json:"max_concurrency,omitempty"`
	ErrorMode      BatchErrorMode `json:"error_mode"`
	MaxItems       int            `json:"max_items,omitempty"`
}

BatchConfig defines an inline item Workflow and its bounded map behavior.

Example 
package main

import (
	"fmt"

	"github.com/rsbin1178/pips/workflow"
)

func main() {
	config := workflow.BatchConfig{
		Mode:           workflow.BatchParallel,
		MaxConcurrency: 4,
		ErrorMode:      workflow.BatchContinueWithNull,
		MaxItems:       100,
	}

	fmt.Println(config.Mode, config.MaxConcurrency, config.ErrorMode)
}

Output:
parallel 4 continue_with_null

type BatchErrorMode 

type BatchErrorMode string

BatchErrorMode controls failed item aggregation. 

const (
	BatchTerminate        BatchErrorMode = "terminate"
	BatchContinueWithNull BatchErrorMode = "continue_with_null"
	BatchRemoveFailed     BatchErrorMode = "remove_failed"
)

Batch item error modes.

type BatchMode 

type BatchMode string

BatchMode controls how Batch items are scheduled. 

const (
	BatchSequential BatchMode = "sequential"
	BatchParallel   BatchMode = "parallel"
)

Batch scheduling modes.

type BatchNode 

type BatchNode struct{}

BatchNode maps an inline Workflow over an input array.

func (BatchNode) Compile 

func (BatchNode) Compile(
	ctx context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (BatchNode) Spec 

func (BatchNode) Spec() NodeTypeSpec

Spec implements NodeType.

type Binding 

type Binding struct {
	Source BindingSource `json:"source"`
	Node   NodeID        `json:"node,omitempty"`
	Port   string        `json:"port,omitempty"`
	Path   []string      `json:"path,omitempty"`
	Value  *Value        `json:"value,omitempty"`
}

Binding selects a literal, Workflow input, prior node output, typed node failure data, or direct Loop variable. Path walks object keys or decimal array indexes after resolving the source value.

type BindingSource 

type BindingSource string

BindingSource identifies where an input value comes from. 

const (
	BindingLiteral       BindingSource = "literal"
	BindingWorkflowInput BindingSource = "workflow_input"
	BindingNodeOutput    BindingSource = "node_output"
	BindingNodeError     BindingSource = "node_error"
	BindingLoopVariable  BindingSource = "loop_variable"
)

Binding sources.

type BreakNode 

type BreakNode struct{}

BreakNode terminates the owning Loop after the current iteration commits.

func (BreakNode) Compile 

func (BreakNode) Compile(
	_ context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (BreakNode) Spec 

func (BreakNode) Spec() NodeTypeSpec

Spec implements NodeType.

type CheckpointStore 

type CheckpointStore interface {
	Get(context.Context, string) ([]byte, bool, error)
	Set(context.Context, string, []byte) error
}

CheckpointStore persists one opaque, versioned Workflow checkpoint by Run ID. Implementations must replace a key atomically and detach returned buffers.

type CompileContext 

type CompileContext interface {
	WorkflowInputs() map[string]PortSchema
	WorkflowOutputs() map[string]PortSchema
	Action(ActionKey, string) (Action, bool)
}

CompileContext exposes Definition contracts and registered Actions during NodeType.Compile without exposing mutable Registry internals. NodeTypes must not retain a CompileContext after Compile returns.

type CompileControlPath 

type CompileControlPath struct {
	From  NodePath
	To    NodePath
	Route string
}

CompileControlPath identifies one control connection in a root or nested Definition. From and To are complete paths through composite boundaries.

type CompileError 

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

CompileError reports one or more ordered issues that prevented Plan construction. Compile aggregates only the earliest failed dependency phase; later phases are not evaluated after their prerequisites fail.

func (*CompileError) Error 

func (e *CompileError) Error() string

Error returns a deterministic summary of every issue.

func (*CompileError) Issues 

func (e *CompileError) Issues() []CompileIssue

Issues returns an independent snapshot of the ordered diagnostics.

func (*CompileError) Unwrap 

func (e *CompileError) Unwrap() []error

Unwrap preserves ErrCompile and any underlying Definition, resolver, or NodeType causes for errors.Is and errors.As.

type CompileIssue 

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

CompileIssue is one immutable, process-local compiler diagnostic. It has no prescribed JSON or persistence format; hosts own versioned projections.

func (CompileIssue) Code 

func (i CompileIssue) Code() CompileIssueCode

Code returns the stable category of i.

func (CompileIssue) ControlPath 

func (i CompileIssue) ControlPath() (CompileControlPath, bool)

ControlPath returns the complete control path when i is path-located.

func (CompileIssue) Location 

func (i CompileIssue) Location() CompileIssueLocation

Location returns the location kind of i.

func (CompileIssue) Message 

func (i CompileIssue) Message() string

Message returns the human-readable detail of i.

func (CompileIssue) NodePath 

func (i CompileIssue) NodePath() (NodePath, bool)

NodePath returns the complete node path when i is node-located.

type CompileIssueCode 

type CompileIssueCode string

CompileIssueCode identifies a stable compile failure category. Messages may become more specific over time; callers should branch on Code instead. 

const (
	CompileIssueInvalidDefinition  CompileIssueCode = "invalid_definition"
	CompileIssueInvalidInterrupt   CompileIssueCode = "invalid_interrupt"
	CompileIssueInvalidReference   CompileIssueCode = "invalid_reference"
	CompileIssueUnknownNodeType    CompileIssueCode = "unknown_node_type"
	CompileIssueInvalidNodeConfig  CompileIssueCode = "invalid_node_config"
	CompileIssueInvalidNodeSpec    CompileIssueCode = "invalid_node_spec"
	CompileIssueInvalidControlPath CompileIssueCode = "invalid_control_path"
	CompileIssueInvalidGraph       CompileIssueCode = "invalid_graph"
	CompileIssueInvalidBinding     CompileIssueCode = "invalid_binding"
)

Compile issue codes.

type CompileIssueLocation 

type CompileIssueLocation string

CompileIssueLocation identifies which location accessor is available on a CompileIssue. 

const (
	CompileLocationDefinition CompileIssueLocation = "definition"
	CompileLocationNode       CompileIssueLocation = "node"
	CompileLocationPath       CompileIssueLocation = "path"
)

Compile issue locations.

type CompileOption 

type CompileOption func(*compileConfig) error

CompileOption configures child Workflow resolution during compilation.

func WithDefinitionResolver 

func WithDefinitionResolver(resolver DefinitionResolver) CompileOption

WithDefinitionResolver configures exact referenced Workflow resolution.

func WithInterruptAfterNodes 

func WithInterruptAfterNodes(paths ...NodePath) CompileOption

WithInterruptAfterNodes pauses a Run after any addressed node succeeds.

func WithInterruptBeforeNodes 

func WithInterruptBeforeNodes(paths ...NodePath) CompileOption

WithInterruptBeforeNodes pauses a Run before any addressed node is invoked.

type CompiledNode 

type CompiledNode interface {
	Spec() NodeSpec
	Invoke(context.Context, NodeInput) (NodeOutput, error)
}

CompiledNode is an immutable, concurrent-safe node executor. Invoke must not retain or mutate NodeInput.Values.

type ConditionConfig 

type ConditionConfig struct {
	Predicate  Predicate `json:"predicate"`
	TrueRoute  string    `json:"true_route"`
	FalseRoute string    `json:"false_route"`
}

ConditionConfig selects named routes for a boolean Predicate result.

type ConditionNode 

type ConditionNode struct{}

ConditionNode chooses one named route using a structured Predicate.

func (ConditionNode) Compile 

func (ConditionNode) Compile(
	_ context.Context,
	_ CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (ConditionNode) Spec 

func (ConditionNode) Spec() NodeTypeSpec

Spec implements NodeType.

type ContinueNode 

type ContinueNode struct{}

ContinueNode terminates the current Loop iteration without exiting the Loop.

func (ContinueNode) Compile 

func (ContinueNode) Compile(
	_ context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (ContinueNode) Spec 

func (ContinueNode) Spec() NodeTypeSpec

Spec implements NodeType.

type ControlEdge 

type ControlEdge struct {
	From NodeRoute `json:"from"`
	To   NodeID    `json:"to"`
}

ControlEdge connects one named source route to a target node. It carries no data; values move only through explicit Bindings.

type Definition 

type Definition struct {
	Schema   string       `json:"schema"`
	ID       DefinitionID `json:"id"`
	Revision Revision     `json:"revision"`
	Name     string       `json:"name"`

	Inputs  map[string]WorkflowInput `json:"inputs"`
	Outputs map[string]OutputBinding `json:"outputs"`
	Nodes   []NodeDefinition         `json:"nodes"`
	Edges   []ControlEdge            `json:"edges"`
	Limits  Limits                   `json:"limits"`
}

Definition is a versioned, serializable Workflow contract.

func DecodeDefinition 

func DecodeDefinition(data []byte) (Definition, error)

DecodeDefinition strictly decodes and validates one Definition snapshot.

func (Definition) Fingerprint 

func (d Definition) Fingerprint() (string, error)

Fingerprint returns a SHA-256 digest of execution-semantic fields. ID, Revision, and Name do not affect it.

func (Definition) MarshalJSON 

func (d Definition) MarshalJSON() ([]byte, error)

MarshalJSON writes the exact wire shape selected by Definition.Schema.

func (*Definition) UnmarshalJSON 

func (d *Definition) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes only the exact wire shape selected by the schema field.

type DefinitionID 

type DefinitionID string

DefinitionID identifies a logical workflow across revisions.

type DefinitionRef 

type DefinitionRef struct {
	ID          DefinitionID `json:"id"`
	Revision    Revision     `json:"revision"`
	Fingerprint string       `json:"fingerprint"`
}

DefinitionRef pins one exact immutable Workflow definition.

Example 
package main

import (
	"fmt"

	"github.com/rsbin1178/pips/workflow"
)

func main() {
	reference := workflow.DefinitionRef{
		ID:          "thumbnail-flow",
		Revision:    "v3",
		Fingerprint: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
	}

	fmt.Println(reference.ID, reference.Revision)
}

Output:
thumbnail-flow v3

type DefinitionResolver 

type DefinitionResolver interface {
	ResolveDefinition(context.Context, DefinitionID, Revision) (Definition, error)
}

DefinitionResolver resolves exact Workflow revisions for compilation. Implementations must be safe for concurrent calls and return detached Definition snapshots.

type DefinitionResolverFunc 

type DefinitionResolverFunc func(context.Context, DefinitionID, Revision) (Definition, error)

DefinitionResolverFunc adapts a function to DefinitionResolver.

func (DefinitionResolverFunc) ResolveDefinition 

func (f DefinitionResolverFunc) ResolveDefinition(
	ctx context.Context,
	id DefinitionID,
	revision Revision,
) (Definition, error)

ResolveDefinition implements DefinitionResolver.

type EndNode 

type EndNode struct{}

EndNode validates and returns Workflow outputs.

func (EndNode) Compile 

func (EndNode) Compile(
	_ context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (EndNode) Spec 

func (EndNode) Spec() NodeTypeSpec

Spec implements NodeType.

type ErrorAction 

type ErrorAction string

ErrorAction controls a node failure after retries are exhausted. 

const (
	ErrorStop                ErrorAction = "stop"
	ErrorRoute               ErrorAction = "route_error"
	ErrorContinueWithDefault ErrorAction = "continue_with_default"
)

Node failure actions. The zero value is equivalent to ErrorStop.

type Event 

type Event struct {
	DefinitionID          DefinitionID
	Revision              Revision
	DefinitionFingerprint string
	PlanFingerprint       string
	RunID                 string
	NodeID                NodeID
	NodeType              NodeTypeKey
	Attempt               int
	Time                  time.Time
	// contains filtered or unexported fields
}

Event is one immutable, process-local Workflow lifecycle increment. It has no JSON wire format; durable consumers must define a versioned projection.

func (Event) MarshalJSON 

func (Event) MarshalJSON() ([]byte, error)

MarshalJSON prevents Event from being mistaken for a stable wire protocol.

func (Event) Payload 

func (e Event) Payload() EventPayload

Payload returns e's sealed semantic payload.

func (Event) Scope 

func (e Event) Scope() []ScopeFrame

Scope returns an independent snapshot of nested execution frames. Root events have an empty scope.

func (Event) Type 

func (e Event) Type() EventType

Type returns the discriminator for e's payload.

func (*Event) UnmarshalJSON 

func (*Event) UnmarshalJSON([]byte) error

UnmarshalJSON rejects an unspecified Event wire representation.

type EventPayload 

type EventPayload interface {
	// contains filtered or unexported methods
}

EventPayload is the sealed union of Workflow Event variants.

type EventSink 

type EventSink func(Event)

EventSink synchronously observes process-local events. A Runner serializes calls to one sink for each Run and applies backpressure. The same sink may be called concurrently by concurrent Runs and must synchronize shared state.

type EventType 

type EventType string

EventType identifies one process-local Workflow event variant. 

const (
	EventRunStarted      EventType = "run_started"
	EventRunCompleted    EventType = "run_completed"
	EventRunFailed       EventType = "run_failed"
	EventRunCanceled     EventType = "run_canceled"
	EventRunInterrupted  EventType = "run_interrupted"
	EventRunResumed      EventType = "run_resumed"
	EventNodeReady       EventType = "node_ready"
	EventNodeStarted     EventType = "node_started"
	EventNodeCompleted   EventType = "node_completed"
	EventNodeFailed      EventType = "node_failed"
	EventNodeSkipped     EventType = "node_skipped"
	EventNodeRetrying    EventType = "node_retrying"
	EventNodeInterrupted EventType = "node_interrupted"
)

Event types.

type ExecutionContext 

type ExecutionContext struct {
	RunID        string
	DefinitionID DefinitionID
	Revision     Revision
	Address      NodeAddress
	Attempt      int
}

ExecutionContext identifies one concrete node invocation. Attempt is the one-based cumulative attempt for Address within the Run, including attempts restored from a checkpoint.

ExecutionContext is correlation metadata, not an external idempotency key. Hosts define idempotency according to their own side-effect boundaries.

func GetExecutionContext 

func GetExecutionContext(ctx context.Context) (ExecutionContext, bool)

GetExecutionContext returns the scheduler-owned identity of the current node invocation. It returns false outside an invocation. The returned Address is detached and may be modified without affecting runtime state or later reads.

type FailureKind 

type FailureKind string

FailureKind is a bounded, payload-free node failure classification. 

const (
	FailureError    FailureKind = "error"
	FailureTimeout  FailureKind = "timeout"
	FailurePanic    FailureKind = "panic"
	FailureCanceled FailureKind = "canceled"
	FailureLimit    FailureKind = "limit"
)

Failure kinds. Action error text is deliberately absent from Events.

type InterruptContext 

type InterruptContext struct {
	ID      string
	Address NodeAddress
	Info    Value
}

InterruptContext describes one targetable dynamic interruption.

type InterruptError 

type InterruptError struct {
	RunID string
	Info  InterruptInfo
}

InterruptError reports a successfully checkpointed, resumable Run.

func (*InterruptError) Error 

func (e *InterruptError) Error() string

Error implements error.

func (*InterruptError) Unwrap 

func (e *InterruptError) Unwrap() error

Unwrap exposes ErrInterrupted for errors.Is.

type InterruptInfo 

type InterruptInfo struct {
	Contexts    []InterruptContext
	BeforeNodes []NodeAddress
	AfterNodes  []NodeAddress
	RerunNodes  []NodeAddress
}

InterruptInfo aggregates every interruption observed at the settled scheduler frontier. Only Contexts are targetable with ResumeTarget.

type Limits 

type Limits struct {
	MaxConcurrency int `json:"max_concurrency"`
	MaxSteps       int `json:"max_steps"`
}

Limits bounds one Run. Both fields must be positive.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits returns conservative in-process Run limits.

type LoopConfig 

type LoopConfig struct {
	Body          Definition     `json:"body"`
	Mode          LoopMode       `json:"mode"`
	Arrays        []string       `json:"arrays,omitempty"`
	Variables     []LoopVariable `json:"variables,omitempty"`
	Outputs       []LoopOutput   `json:"outputs,omitempty"`
	MaxIterations int            `json:"max_iterations,omitempty"`
}

LoopConfig defines an inline sequential Loop body.

type LoopMode 

type LoopMode string

LoopMode controls how a Loop determines its iteration count. 

const (
	LoopArray    LoopMode = "array"
	LoopCount    LoopMode = "count"
	LoopInfinite LoopMode = "infinite"
)

Loop iteration modes.

type LoopNode 

type LoopNode struct{}

LoopNode repeatedly executes an inline Workflow with transactional local variables.

Example 
integerSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"integer"}`))
countSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"integer","minimum":1}`))
stringSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"string"}`))
ready := workflow.MustValueOf("ready")

conditionConfig, _ := json.Marshal(workflow.ConditionConfig{
	Predicate: workflow.Predicate{
		Op: workflow.PredicateEqual, Input: "state", Value: &ready,
	},
	TrueRoute: "stop", FalseRoute: "next",
})
setConfig, _ := json.Marshal(workflow.SetVariableConfig{
	Assignments: []workflow.SetVariableAssignment{{Target: "state", Input: "next"}},
})
body := workflow.Definition{
	Schema: workflow.SchemaV1Alpha1, ID: "feedback-body", Revision: "v1", Name: "Feedback Body",
	Inputs:  map[string]workflow.WorkflowInput{"index": {Schema: integerSchema, Required: true}},
	Outputs: map[string]workflow.OutputBinding{},
	Nodes: []workflow.NodeDefinition{
		{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
		{
			ID: "condition", Type: workflow.NodeTypeCondition, Version: workflow.BuiltinNodeVersion,
			Config: conditionConfig,
			Inputs: map[string]workflow.Binding{
				"state": {Source: workflow.BindingLoopVariable, Port: "state"},
			},
		},
		{
			ID: "set", Type: workflow.NodeTypeSetVariable, Version: workflow.BuiltinNodeVersion,
			Config: setConfig,
			Inputs: map[string]workflow.Binding{
				"next": {Source: workflow.BindingLiteral, Value: &ready},
			},
		},
		{ID: "break", Type: workflow.NodeTypeBreak, Version: workflow.BuiltinNodeVersion},
		{ID: "continue", Type: workflow.NodeTypeContinue, Version: workflow.BuiltinNodeVersion},
		{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
	},
	Edges: []workflow.ControlEdge{
		edge("start", workflow.RouteSuccess, "condition"),
		edge("condition", "stop", "break"),
		edge("condition", "next", "set"),
		edge("set", workflow.RouteSuccess, "continue"),
		edge("break", workflow.RouteSuccess, "end"),
		edge("continue", workflow.RouteSuccess, "end"),
	},
	Limits: workflow.DefaultLimits(),
}
loopConfig, _ := json.Marshal(workflow.LoopConfig{
	Body: body, Mode: workflow.LoopCount,
	Variables: []workflow.LoopVariable{{Name: "state", Schema: stringSchema}},
	Outputs: []workflow.LoopOutput{
		{Name: "final", Source: workflow.LoopOutputVariable, Port: "state"},
	},
	MaxIterations: 10,
})
definition := workflow.Definition{
	Schema: workflow.SchemaV1Alpha1, ID: "feedback", Revision: "v1", Name: "Feedback",
	Inputs: map[string]workflow.WorkflowInput{
		"count": {Schema: countSchema, Required: true}, "state": {Schema: stringSchema, Required: true},
	},
	Outputs: map[string]workflow.OutputBinding{
		"final": nodeOutput(stringSchema, "loop", "final"),
	},
	Nodes: []workflow.NodeDefinition{
		{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
		{
			ID: "loop", Type: workflow.NodeTypeLoop, Version: workflow.BuiltinNodeVersion,
			Config: loopConfig,
			Inputs: map[string]workflow.Binding{
				"count": workflowInput("count"), "state": workflowInput("state"),
			},
		},
		{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
	},
	Edges: []workflow.ControlEdge{
		edge("start", workflow.RouteSuccess, "loop"),
		edge("loop", workflow.RouteSuccess, "end"),
	},
	Limits: workflow.DefaultLimits(),
}

registry, _ := workflow.NewDefaultRegistry()
plan, _ := workflow.Compile(context.Background(), definition, registry)
runner, _ := workflow.NewRunner()
result, _ := runner.Run(context.Background(), plan, map[string]workflow.Value{
	"count": workflow.MustValueOf(5), "state": workflow.MustValueOf("pending"),
})

fmt.Println(result.Outputs["final"].String())

Output:
"ready"

func (LoopNode) Compile 

func (LoopNode) Compile(
	ctx context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (LoopNode) Spec 

func (LoopNode) Spec() NodeTypeSpec

Spec implements NodeType.

type LoopOutput 

type LoopOutput struct {
	Name   string           `json:"name"`
	Source LoopOutputSource `json:"source"`
	Port   string           `json:"port"`
}

LoopOutput projects a body output or final Loop variable onto the Loop node.

type LoopOutputSource 

type LoopOutputSource string

LoopOutputSource identifies the value projected by a Loop output. 

const (
	LoopOutputBody     LoopOutputSource = "body_output"
	LoopOutputVariable LoopOutputSource = "loop_variable"
)

Loop output sources.

type LoopVariable 

type LoopVariable struct {
	Name   string     `json:"name"`
	Schema PortSchema `json:"schema"`
}

LoopVariable declares one Loop-local mutable value and its schema.

type MergeConfig 

type MergeConfig struct {
	Mode    MergeMode                    `json:"mode"`
	Outputs map[string]MergeOutputConfig `json:"outputs"`
}

MergeConfig defines explicit output mappings for one Merge mode.

type MergeMode 

type MergeMode string

MergeMode selects mutually exclusive or parallel fan-in behavior. 

const (
	MergeExclusive MergeMode = "exclusive"
	MergeParallel  MergeMode = "parallel"
)

Merge modes.

type MergeNode 

type MergeNode struct{}

MergeNode joins mutually exclusive branches or wait-all parallel inputs.

func (MergeNode) Compile 

func (MergeNode) Compile(
	_ context.Context,
	_ CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (MergeNode) Spec 

func (MergeNode) Spec() NodeTypeSpec

Spec implements NodeType.

type MergeOutputConfig 

type MergeOutputConfig struct {
	Schema  PortSchema `json:"schema"`
	Sources []string   `json:"sources"`
}

MergeOutputConfig maps one output to explicitly named node input candidates.

type NodeAddress 

type NodeAddress struct {
	NodeID NodeID
	Scope  []ScopeFrame
}

NodeAddress identifies one node invocation inside its composite scope.

type NodeCompleted 

type NodeCompleted struct{}

NodeCompleted closes a successful invocation attempt.

type NodeDebugExecution 

type NodeDebugExecution struct {
	Address      NodeAddress
	NodeRun      NodeRun
	Inputs       map[string]Value
	Outputs      map[string]Value
	Route        string
	ErrorMessage string
}

NodeDebugExecution is one detached node execution record. ErrorMessage may contain sensitive Action error text and must not be treated as an Event.

type NodeDebugPlan 

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

NodeDebugPlan is an immutable, concurrent-safe isolated node execution plan.

func PrepareNodeDebug 

func PrepareNodeDebug(source *Plan, target NodePath) (*NodeDebugPlan, error)

PrepareNodeDebug derives an immutable isolated executable for target without invoking a node or mutating source.

func (*NodeDebugPlan) Fingerprint 

func (p *NodeDebugPlan) Fingerprint() string

Fingerprint returns the derived isolated executable identity.

func (*NodeDebugPlan) SourcePlanFingerprint 

func (p *NodeDebugPlan) SourcePlanFingerprint() string

SourcePlanFingerprint returns the exact source Plan identity.

func (*NodeDebugPlan) Spec 

func (p *NodeDebugPlan) Spec() NodeDebugSpec

Spec returns a detached explicit input and selected output/route contract.

func (*NodeDebugPlan) Target 

func (p *NodeDebugPlan) Target() NodePath

Target returns the original static path of the selected node.

type NodeDebugResult 

type NodeDebugResult struct {
	RunID           string
	Target          NodePath
	Status          RunStatus
	Execution       NodeDebugExecution
	InnerExecutions []NodeDebugExecution
	StartedAt       time.Time
	EndedAt         time.Time
	Interruption    *InterruptInfo
}

NodeDebugResult is the detached result of one isolated node trial. Inputs, outputs, and ErrorMessage may contain sensitive application data.

Example (Composite) 
stringSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"string"}`))
integerSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"integer"}`))
itemsSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"array","items":{"type":"string"}}`))
action := &fakeAction{
	spec: actionSpec("map-item", map[string]workflow.PortSchema{
		"item": stringSchema, "index": integerSchema,
	}, map[string]workflow.PortSchema{"result": stringSchema}),
	run: func(_ context.Context, input workflow.ActionInput) (workflow.ActionOutput, error) {
		return workflow.ActionOutput{Values: map[string]workflow.Value{
			"result": input.Values["item"],
		}}, nil
	},
}
body := workflow.Definition{
	Schema: workflow.SchemaV1Alpha1, ID: "map-body", Revision: "v1", Name: "Map Body",
	Inputs: map[string]workflow.WorkflowInput{"item": {Schema: stringSchema, Required: true}, "index": {Schema: integerSchema, Required: true}},
	Outputs: map[string]workflow.OutputBinding{
		"result": nodeOutput(stringSchema, "map", "result"),
	},
	Nodes: []workflow.NodeDefinition{
		{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
		{
			ID: "map", Type: workflow.NodeTypeAction, Version: workflow.BuiltinNodeVersion,
			Config: json.RawMessage(`{"action":"map-item","version":"v1"}`),
			Inputs: map[string]workflow.Binding{
				"item": workflowInput("item"), "index": workflowInput("index"),
			},
		},
		{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
	},
	Edges: []workflow.ControlEdge{
		edge("start", workflow.RouteSuccess, "map"),
		edge("map", workflow.RouteSuccess, "end"),
	},
	Limits: workflow.DefaultLimits(),
}
config, _ := json.Marshal(workflow.BatchConfig{
	Body: body, ResultOutput: "result", Mode: workflow.BatchParallel,
	ErrorMode: workflow.BatchTerminate, MaxItems: 10, MaxConcurrency: 2,
})
definition := workflow.Definition{
	Schema: workflow.SchemaV1Alpha1, ID: "batch-preview", Revision: "v1", Name: "Batch Preview",
	Inputs: map[string]workflow.WorkflowInput{"items": {Schema: itemsSchema, Required: true}},
	Outputs: map[string]workflow.OutputBinding{
		"results": nodeOutput(itemsSchema, "batch", "results"),
	},
	Nodes: []workflow.NodeDefinition{
		{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
		{
			ID: "batch", Type: workflow.NodeTypeBatch, Version: workflow.BuiltinNodeVersion,
			Config: config, Inputs: map[string]workflow.Binding{"items": workflowInput("items")},
		},
		{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
	},
	Edges: []workflow.ControlEdge{
		edge("start", workflow.RouteSuccess, "batch"),
		edge("batch", workflow.RouteSuccess, "end"),
	},
	Limits: workflow.DefaultLimits(),
}
registry, _ := workflow.NewDefaultRegistry(action)
plan, _ := workflow.Compile(context.Background(), definition, registry)
debugPlan, _ := workflow.PrepareNodeDebug(plan, workflow.NewNodePath("batch"))
runner, _ := workflow.NewRunner()
result, _ := runner.DebugNode(context.Background(), debugPlan, map[string]workflow.Value{
	"items": workflow.MustValueOf([]string{"a", "b"}),
})

fmt.Println(result.Execution.Outputs["results"].String())

for _, inner := range result.InnerExecutions {
	fmt.Println(inner.Address.NodeID, inner.Address.Scope[0].Index)
}

Output:
["a","b"]
map 0
map 1

type NodeDebugSpec 

type NodeDebugSpec struct {
	Inputs         map[string]PortSchema
	OptionalInputs []string
	Outputs        map[string]PortSchema
	Routes         []string
}

NodeDebugSpec is the explicit caller contract for one isolated node trial. Literal-bound inputs are retained by the prepared plan and are not exposed.

type NodeDefinition 

type NodeDefinition struct {
	ID      NodeID             `json:"id"`
	Name    string             `json:"name,omitempty"`
	Type    NodeTypeKey        `json:"type"`
	Version string             `json:"version"`
	Config  json.RawMessage    `json:"config,omitempty"`
	Inputs  map[string]Binding `json:"inputs,omitempty"`
	Policy  NodePolicy         `json:"policy,omitzero"`
}

NodeDefinition is one version-pinned node instance.

type NodeExecution 

type NodeExecution struct {
	DefinitionID          DefinitionID
	Revision              Revision
	DefinitionFingerprint string
	PlanFingerprint       string
	RunID                 string
	Address               NodeAddress
	NodeRun               NodeRun
	Inputs                map[string]Value
	Outputs               map[string]Value
	Route                 string
	ErrorMessage          string
}

NodeExecution is one detached, storage-neutral snapshot of an actual node invocation. RunID and Address form its logical upsert identity. Inputs, outputs, and error text may contain sensitive application data; hosts own persistence, redaction, encryption, retention, and transport schemas.

func (NodeExecution) MarshalJSON 

func (NodeExecution) MarshalJSON() ([]byte, error)

MarshalJSON rejects an accidental unversioned wire format for sensitive process-local execution data.

func (*NodeExecution) UnmarshalJSON 

func (*NodeExecution) UnmarshalJSON([]byte) error

UnmarshalJSON rejects an accidental unversioned wire format for sensitive process-local execution data.

type NodeExecutionRecorder 

type NodeExecutionRecorder func(context.Context, NodeExecution)

NodeExecutionRecorder synchronously receives running and effective outcome snapshots for ordinary Run and Resume operations. Implementations handle their own persistence errors and timeouts and must not panic.

Calls are serialized within one Run. The same recorder may be called concurrently by separate Runs.

type NodeFailed 

type NodeFailed struct {
	Kind FailureKind
}

NodeFailed closes a failed invocation attempt without retaining error text.

type NodeID 

type NodeID string

NodeID identifies a node inside one Definition.

type NodeInput 

type NodeInput struct {
	Values map[string]Value
	// contains filtered or unexported fields
}

NodeInput contains resolved immutable input values for one invocation.

type NodeInterrupted 

type NodeInterrupted struct{}

NodeInterrupted reports an invocation suspended by a dynamic or descendant interruption.

type NodeOutput 

type NodeOutput struct {
	Values map[string]Value
	Route  string
}

NodeOutput contains immutable output values and the selected control route.

type NodePath 

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

NodePath identifies one node through zero or more composite node boundaries. NewNodePath and Nodes detach their slices so a path is safe to reuse.

func NewNodePath 

func NewNodePath(nodes ...NodeID) NodePath

NewNodePath creates an immutable node path. Compile rejects an empty path or an invalid or unresolved segment.

func (NodePath) Nodes 

func (p NodePath) Nodes() []NodeID

Nodes returns an independent snapshot of the path segments.

type NodePolicy 

type NodePolicy struct {
	TimeoutMilli   int64            `json:"timeout_ms,omitempty"`
	Retry          RetryPolicy      `json:"retry,omitzero"`
	Error          ErrorAction      `json:"error,omitempty"`
	DefaultOutputs map[string]Value `json:"default_outputs,omitempty"`
}

NodePolicy bounds one node invocation and defines its explicit error path.

type NodeReady 

type NodeReady struct{}

NodeReady reports that every incoming control edge has resolved and at least one was taken.

type NodeRetrying 

type NodeRetrying struct {
	NextAttempt int
}

NodeRetrying reports that another bounded attempt will start.

type NodeRoute 

type NodeRoute struct {
	Node  NodeID `json:"node"`
	Route string `json:"route"`
}

NodeRoute selects one named control route from a node.

type NodeRun 

type NodeRun struct {
	ID        NodeID
	Type      NodeTypeKey
	Status    NodeStatus
	Attempts  int
	Failure   FailureKind
	StartedAt time.Time
	EndedAt   time.Time
}

NodeRun is an immutable node execution summary.

type NodeSkipped 

type NodeSkipped struct{}

NodeSkipped reports that every incoming control edge was skipped.

type NodeSpec 

type NodeSpec struct {
	Inputs  map[string]PortSchema
	Outputs map[string]PortSchema
	Routes  []string
}

NodeSpec is the immutable input, output, and route contract produced while compiling one node instance.

type NodeStarted 

type NodeStarted struct{}

NodeStarted opens one invocation attempt.

type NodeStatus 

type NodeStatus string

NodeStatus identifies one node lifecycle state. 

const (
	NodeStatusPending     NodeStatus = "pending"
	NodeStatusReady       NodeStatus = "ready"
	NodeStatusRunning     NodeStatus = "running"
	NodeStatusSucceeded   NodeStatus = "succeeded"
	NodeStatusException   NodeStatus = "exception"
	NodeStatusFailed      NodeStatus = "failed"
	NodeStatusSkipped     NodeStatus = "skipped"
	NodeStatusInterrupted NodeStatus = "interrupted"
)

Node states.

type NodeType 

type NodeType interface {
	Spec() NodeTypeSpec
	Compile(context.Context, CompileContext, NodeDefinition) (CompiledNode, error)
}

NodeType compiles one trusted, application-linked node implementation.

func BuiltinNodeTypes 

func BuiltinNodeTypes() []NodeType

BuiltinNodeTypes returns independent registrations for the built-in Workflow nodes.

type NodeTypeKey 

type NodeTypeKey string

NodeTypeKey identifies a registered node implementation family. 

const (
	// BuiltinNodeVersion is the exact version of all first-version built-ins.
	BuiltinNodeVersion = "v1"
	// MergeNodeVersionV2 selects the first non-null candidate in source order.
	MergeNodeVersionV2 = "v2"

	// NodeTypeStart identifies [StartNode].
	NodeTypeStart NodeTypeKey = "start"
	// NodeTypeEnd identifies [EndNode].
	NodeTypeEnd NodeTypeKey = "end"
	// NodeTypeAction identifies [ActionNode].
	NodeTypeAction NodeTypeKey = "action"
	// NodeTypeCondition identifies [ConditionNode].
	NodeTypeCondition NodeTypeKey = "condition"
	// NodeTypeMerge identifies [MergeNode].
	NodeTypeMerge NodeTypeKey = "merge"
	// NodeTypeSelector identifies [SelectorNode].
	NodeTypeSelector NodeTypeKey = "selector"
	// NodeTypeSubWorkflow identifies [SubWorkflowNode].
	NodeTypeSubWorkflow NodeTypeKey = "sub_workflow"
	// NodeTypeBatch identifies [BatchNode].
	NodeTypeBatch NodeTypeKey = "batch"
	// NodeTypeLoop identifies [LoopNode].
	NodeTypeLoop NodeTypeKey = "loop"
	// NodeTypeBreak identifies [BreakNode].
	NodeTypeBreak NodeTypeKey = "break"
	// NodeTypeContinue identifies [ContinueNode].
	NodeTypeContinue NodeTypeKey = "continue"
	// NodeTypeSetVariable identifies [SetVariableNode].
	NodeTypeSetVariable NodeTypeKey = "set_variable"

	// RouteSuccess is the normal completion route.
	RouteSuccess = "success"
	// RouteError is selected by ErrorRoute after invocation failure.
	RouteError = "error"
)

type NodeTypeSpec 

type NodeTypeSpec struct {
	Key         NodeTypeKey
	Version     string
	DisplayName string
}

NodeTypeSpec identifies one registered node implementation.

type OutputBinding 

type OutputBinding struct {
	Schema  PortSchema `json:"schema"`
	Binding Binding    `json:"binding"`
}

OutputBinding declares one Workflow output contract and its value source.

type PartialDataOrigin 

type PartialDataOrigin string

PartialDataOrigin identifies how one node obtained its effective result in a Partial Run. Its zero value means no result was materialized or executed. 

const (
	PartialDataExecuted PartialDataOrigin = "executed"
	PartialDataPinned   PartialDataOrigin = "pinned"
	PartialDataReused   PartialDataOrigin = "reused"
)

Partial Run data origins.

type PartialNodeData 

type PartialNodeData struct {
	Outputs map[string]Value
	Route   string
}

PartialNodeData is one reusable successful node result.

type PartialNodeRun 

type PartialNodeRun struct {
	NodeRun NodeRun
	Origin  PartialDataOrigin
}

PartialNodeRun combines one node summary with its Partial Run data origin.

type PartialRunData 

type PartialRunData struct {
	RunID                 string
	DefinitionID          DefinitionID
	Revision              Revision
	SourcePlanFingerprint string
	Nodes                 map[NodeID]PartialNodeData
}

PartialRunData is a detached, host-persistable snapshot produced by a Partial Run. It is prior execution data, not a resumable checkpoint.

type PartialRunInput 

type PartialRunInput struct {
	Inputs   map[string]Value
	Previous *PartialRunData
	Pins     PinData
	Dirty    []NodeID
}

PartialRunInput supplies Workflow inputs and optional host-owned development data for Runner.RunPartial.

type PartialRunResult 

type PartialRunResult struct {
	RunID                 string
	Destination           NodeID
	SourcePlanFingerprint string
	PlanFingerprint       string
	Status                RunStatus
	Outputs               map[string]Value
	Nodes                 map[NodeID]PartialNodeRun
	Data                  PartialRunData
	StartedAt             time.Time
	EndedAt               time.Time
	Interruption          *InterruptInfo
}

PartialRunResult is a detached snapshot of one terminal or interrupted Partial Run. Data can be persisted by the host and supplied as Previous to a later Partial Run.

type PinData 

type PinData map[NodeID]map[string]Value

PinData maps root node IDs to explicit successful output snapshots used by Runner.RunPartial. Pins are development data and are never consulted by Runner.Run or Runner.Resume.

type Plan 

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

Plan is an immutable, concurrent-safe compiled Workflow execution plan.

func Compile 

func Compile(
	ctx context.Context,
	definition Definition,
	registry *Registry,
	options ...CompileOption,
) (*Plan, error)

Compile validates definition against registry and returns an immutable Plan.

func (*Plan) DefinitionFingerprint 

func (p *Plan) DefinitionFingerprint() string

DefinitionFingerprint returns the semantic Definition fingerprint.

func (*Plan) DefinitionID 

func (p *Plan) DefinitionID() DefinitionID

DefinitionID returns the logical Workflow identity pinned by p.

func (*Plan) Fingerprint 

func (p *Plan) Fingerprint() string

Fingerprint returns the Definition and referenced execution-contract identity.

func (*Plan) RegistryFingerprint 

func (p *Plan) RegistryFingerprint() string

RegistryFingerprint returns the complete source Registry contract fingerprint.

func (*Plan) Revision 

func (p *Plan) Revision() Revision

Revision returns the exact Definition revision pinned by p.

type PortSchema 

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

PortSchema is an immutable JSON Schema draft 2020-12 contract. External references are rejected because schemas are resolved without a loader.

func ParsePortSchema 

func ParsePortSchema(data []byte) (PortSchema, error)

ParsePortSchema validates, resolves, and snapshots a JSON Schema.

func SchemaFor 

func SchemaFor[T any]() (PortSchema, error)

SchemaFor derives a PortSchema from T using encoding/json field names.

func (PortSchema) Equal 

func (s PortSchema) Equal(other PortSchema) bool

Equal reports whether two schemas have the same canonical representation.

func (PortSchema) Fingerprint 

func (s PortSchema) Fingerprint() string

Fingerprint returns the SHA-256 digest of the canonical schema.

func (PortSchema) IsValid 

func (s PortSchema) IsValid() bool

IsValid reports whether s contains a resolved schema.

func (PortSchema) MarshalJSON 

func (s PortSchema) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (PortSchema) RawJSON 

func (s PortSchema) RawJSON() []byte

RawJSON returns an independent canonical JSON representation.

func (*PortSchema) UnmarshalJSON 

func (s *PortSchema) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler.

func (PortSchema) Validate 

func (s PortSchema) Validate(value Value) (returnErr error)

Validate checks value against s.

type Predicate 

type Predicate struct {
	Op     PredicateOp `json:"op"`
	Input  string      `json:"input,omitempty"`
	Path   []string    `json:"path,omitempty"`
	Value  *Value      `json:"value,omitempty"`
	Values []Value     `json:"values,omitempty"`
	Args   []Predicate `json:"args,omitempty"`
}

Predicate is a bounded, serializable Condition expression. Leaf operations read Input and Path; boolean operations contain Args.

type PredicateOp 

type PredicateOp string

PredicateOp identifies one safe Condition predicate operation. 

const (
	PredicateExists         PredicateOp = "exists"
	PredicateEqual          PredicateOp = "eq"
	PredicateNotEqual       PredicateOp = "ne"
	PredicateGreater        PredicateOp = "gt"
	PredicateGreaterOrEqual PredicateOp = "gte"
	PredicateLess           PredicateOp = "lt"
	PredicateLessOrEqual    PredicateOp = "lte"
	PredicateIn             PredicateOp = "in"
	PredicateAll            PredicateOp = "all"
	PredicateAny            PredicateOp = "any"
	PredicateNot            PredicateOp = "not"
)

Condition predicate operations.

type Registry 

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

Registry is an immutable, concurrent-safe snapshot of NodeTypes and Actions.

func NewDefaultRegistry 

func NewDefaultRegistry(actions ...Action) (*Registry, error)

NewDefaultRegistry creates a Registry containing the built-in node types and the supplied Actions.

func NewRegistry 

func NewRegistry(nodeTypes []NodeType, actions []Action) (*Registry, error)

NewRegistry validates and freezes explicitly supplied NodeTypes and Actions.

func (*Registry) Action 

func (r *Registry) Action(key ActionKey, version string) (Action, bool)

Action resolves one exact Action implementation version.

func (*Registry) Fingerprint 

func (r *Registry) Fingerprint() string

Fingerprint identifies the complete NodeType and Action contract catalog.

func (*Registry) NodeType 

func (r *Registry) NodeType(key NodeTypeKey, version string) (NodeType, bool)

NodeType resolves one exact node implementation version.

type ResumeTarget 

type ResumeTarget struct {
	InterruptID string
	Data        Value
}

ResumeTarget supplies optional data to one dynamic interruption ID.

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts  int   `json:"max_attempts,omitempty"`
	BackoffMilli int64 `json:"backoff_ms,omitempty"`
}

RetryPolicy bounds retries for one node. Zero MaxAttempts means one attempt and therefore no retry.

type Revision 

type Revision string

Revision identifies one immutable Definition revision.

type RunCanceled 

type RunCanceled struct{}

RunCanceled closes a canceled Run.

type RunCompleted 

type RunCompleted struct{}

RunCompleted closes a succeeded or partial-succeeded Run.

type RunError 

type RunError struct {
	NodeID  NodeID
	Attempt int
	Err     error
}

RunError identifies the node and attempt that terminated a Run.

func (*RunError) Error 

func (e *RunError) Error() string

Error implements error.

func (*RunError) Unwrap 

func (e *RunError) Unwrap() []error

Unwrap exposes both ErrRun and the invocation cause.

type RunFailed 

type RunFailed struct{}

RunFailed closes a failed Run.

type RunIDSource 

type RunIDSource func(time.Time) (string, error)

RunIDSource creates one Run ID from the Run start time.

type RunInterruptOption 

type RunInterruptOption func(*runInterruptOptions)

RunInterruptOption configures one external host interruption request.

func WithRunInterruptTimeout 

func WithRunInterruptTimeout(timeout time.Duration) RunInterruptOption

WithRunInterruptTimeout bounds how long the scheduler waits for in-flight invocations to settle. Zero and negative durations request immediate cooperative cancellation of unfinished leaf invocations.

type RunInterrupted 

type RunInterrupted struct{}

RunInterrupted reports that a resumable checkpoint was saved.

type RunResult 

type RunResult struct {
	RunID        string
	Status       RunStatus
	Outputs      map[string]Value
	Nodes        map[NodeID]NodeRun
	StartedAt    time.Time
	EndedAt      time.Time
	Interruption *InterruptInfo
}

RunResult is a detached snapshot of one terminal or interrupted Run. An interrupted result keeps EndedAt zero and exposes its resumable frontier in Interruption. Callers may mutate its maps without affecting runtime state.

type RunResumed 

type RunResumed struct{}

RunResumed reports that a checkpoint was validated and execution continued.

type RunStarted 

type RunStarted struct{}

RunStarted opens one Run.

type RunStatus 

type RunStatus string

RunStatus identifies one Run lifecycle state. 

const (
	RunStatusPending          RunStatus = "pending"
	RunStatusRunning          RunStatus = "running"
	RunStatusSucceeded        RunStatus = "succeeded"
	RunStatusPartialSucceeded RunStatus = "partial-succeeded"
	RunStatusFailed           RunStatus = "failed"
	RunStatusCanceled         RunStatus = "canceled"
	RunStatusInterrupted      RunStatus = "interrupted"
)

Run states.

type Runner 

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

Runner synchronously executes immutable Plans in process.

func NewRunner 

func NewRunner(options ...RunnerOption) (*Runner, error)

NewRunner creates a Runner with cryptographically random Run IDs.

func (*Runner) DebugNode 

func (r *Runner) DebugNode(
	ctx context.Context,
	debugPlan *NodeDebugPlan,
	inputs map[string]Value,
) (NodeDebugResult, error)

DebugNode executes only the selected node boundary through the ordinary Workflow scheduler. Registered Actions are real and may have side effects.

Example 
stringSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"string"}`))
action := &fakeAction{
	spec: actionSpec("preview", map[string]workflow.PortSchema{
		"prompt": stringSchema,
	}, map[string]workflow.PortSchema{"result": stringSchema}),
	run: func(_ context.Context, input workflow.ActionInput) (workflow.ActionOutput, error) {
		return workflow.ActionOutput{Values: map[string]workflow.Value{
			"result": input.Values["prompt"],
		}}, nil
	},
}
registry, _ := workflow.NewDefaultRegistry(action)
definition := workflow.Definition{
	Schema: workflow.SchemaV1Alpha1, ID: "preview-flow", Revision: "v1", Name: "Preview Flow",
	Inputs: map[string]workflow.WorkflowInput{"prompt": {Schema: stringSchema, Required: true}},
	Outputs: map[string]workflow.OutputBinding{
		"result": nodeOutput(stringSchema, "preview", "result"),
	},
	Nodes: []workflow.NodeDefinition{
		{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
		{
			ID: "preview", Type: workflow.NodeTypeAction, Version: workflow.BuiltinNodeVersion,
			Config: json.RawMessage(`{"action":"preview","version":"v1"}`),
			Inputs: map[string]workflow.Binding{"prompt": workflowInput("prompt")},
		},
		{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
	},
	Edges: []workflow.ControlEdge{
		edge("start", workflow.RouteSuccess, "preview"),
		edge("preview", workflow.RouteSuccess, "end"),
	},
	Limits: workflow.DefaultLimits(),
}
plan, _ := workflow.Compile(context.Background(), definition, registry)
debugPlan, _ := workflow.PrepareNodeDebug(plan, workflow.NewNodePath("preview"))
runner, _ := workflow.NewRunner(
	workflow.WithRunIDSource(func(time.Time) (string, error) { return "preview-run", nil }),
)
result, _ := runner.DebugNode(context.Background(), debugPlan, map[string]workflow.Value{
	"prompt": workflow.MustValueOf("paper city"),
})

fmt.Println(result.RunID, result.Status, result.Execution.Route)
fmt.Println(result.Execution.Outputs["result"].String())

Output:
preview-run succeeded success
"paper city"

func (*Runner) Resume 

func (r *Runner) Resume(
	ctx context.Context,
	plan *Plan,
	runID string,
	targets []ResumeTarget,
) (RunResult, error)

Resume validates and restores one interrupted Run from its CheckpointStore. Dynamic targets are validated as one set before any node is invoked.

func (*Runner) ResumeNodeDebug 

func (r *Runner) ResumeNodeDebug(
	ctx context.Context,
	debugPlan *NodeDebugPlan,
	runID string,
	targets []ResumeTarget,
) (NodeDebugResult, error)

ResumeNodeDebug resumes one interrupted isolated node trial. The exact same NodeDebugPlan and Run ID are required before any node can be invoked.

Example 
action := &fakeAction{
	spec: actionSpec("resume-preview", map[string]workflow.PortSchema{}, map[string]workflow.PortSchema{}),
}
registry, _ := workflow.NewDefaultRegistry(action)
definition := workflow.Definition{
	Schema: workflow.SchemaV1Alpha1, ID: "resume-preview", Revision: "v1", Name: "Resume Preview",
	Inputs: map[string]workflow.WorkflowInput{}, Outputs: map[string]workflow.OutputBinding{},
	Nodes: []workflow.NodeDefinition{
		{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
		{
			ID: "work", Type: workflow.NodeTypeAction, Version: workflow.BuiltinNodeVersion,
			Config: json.RawMessage(`{"action":"resume-preview","version":"v1"}`),
		},
		{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
	},
	Edges: []workflow.ControlEdge{
		edge("start", workflow.RouteSuccess, "work"),
		edge("work", workflow.RouteSuccess, "end"),
	},
	Limits: workflow.DefaultLimits(),
}
plan, _ := workflow.Compile(
	context.Background(), definition, registry,
	workflow.WithInterruptBeforeNodes(workflow.NewNodePath("work")),
)
debugPlan, _ := workflow.PrepareNodeDebug(plan, workflow.NewNodePath("work"))
store := &memoryCheckpointStore{}
runner, _ := workflow.NewRunner(
	workflow.WithCheckpointStore(store),
	workflow.WithRunIDSource(func(time.Time) (string, error) { return "resume-preview-run", nil }),
)
paused, _ := runner.DebugNode(context.Background(), debugPlan, map[string]workflow.Value{})
completed, _ := runner.ResumeNodeDebug(
	context.Background(), debugPlan, paused.RunID, nil,
)

fmt.Println(paused.Status, completed.Status, completed.RunID)

Output:
interrupted succeeded resume-preview-run

func (*Runner) ResumePartial 

func (r *Runner) ResumePartial(
	ctx context.Context,
	source *Plan,
	destination NodeID,
	runID string,
	targets []ResumeTarget,
) (PartialRunResult, error)

ResumePartial resumes one interrupted Partial Run. The exact same source Plan, destination, Run ID, and outstanding dynamic targets are required before any node can be invoked.

func (*Runner) Run 

func (r *Runner) Run(
	ctx context.Context,
	plan *Plan,
	inputs map[string]Value,
) (RunResult, error)

Run validates inputs, executes plan, and waits for every Runner-owned goroutine before returning. Timeouts and cancellation are cooperative: Actions must return when their context is done.

Example 
package main

import (
	"context"
	"fmt"
	"time"

	"github.com/rsbin1178/pips/workflow"
)

func edge(from workflow.NodeID, route string, to workflow.NodeID) workflow.ControlEdge {
	return workflow.ControlEdge{From: workflow.NodeRoute{Node: from, Route: route}, To: to}
}

func main() {
	stringSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"string"}`))
	definition := workflow.Definition{
		Schema:   workflow.SchemaV1Alpha1,
		ID:       "example",
		Revision: "v1",
		Name:     "Example",
		Inputs:   map[string]workflow.WorkflowInput{"input": {Schema: stringSchema, Required: true}},
		Outputs: map[string]workflow.OutputBinding{
			"result": {
				Schema: stringSchema,
				Binding: workflow.Binding{
					Source: workflow.BindingWorkflowInput,
					Port:   "input",
				},
			},
		},
		Nodes: []workflow.NodeDefinition{
			{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
			{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
		},
		Edges: []workflow.ControlEdge{
			edge("start", workflow.RouteSuccess, "end"),
		},
		Limits: workflow.DefaultLimits(),
	}

	registry, _ := workflow.NewDefaultRegistry()
	plan, _ := workflow.Compile(context.Background(), definition, registry)
	runner, _ := workflow.NewRunner(
		workflow.WithRunIDSource(func(time.Time) (string, error) { return "example-run", nil }),
	)
	result, _ := runner.Run(
		context.Background(),
		plan,
		map[string]workflow.Value{"input": workflow.MustValueOf("hello")},
	)

	fmt.Println(result.Status, result.Outputs["result"].String())
}

Output:
succeeded "hello"

func (*Runner) RunPartial 

func (r *Runner) RunPartial(
	ctx context.Context,
	source *Plan,
	destination NodeID,
	input PartialRunInput,
) (PartialRunResult, error)

RunPartial executes only the root dependency slice needed to settle destination. Previous results and Pins are development data boundaries; registered Actions that do execute are real and may have side effects.

Example 
package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/rsbin1178/pips/workflow"
)

type partialRunExampleAction struct {
	schema workflow.PortSchema
}

func (a partialRunExampleAction) Spec() workflow.ActionSpec {
	return workflow.ActionSpec{
		Key: "render-preview", Version: "v1",
		Inputs:  map[string]workflow.PortSchema{"prompt": a.schema},
		Outputs: map[string]workflow.PortSchema{"image": a.schema},
	}
}

func (a partialRunExampleAction) Run(
	_ context.Context,
	input workflow.ActionInput,
) (workflow.ActionOutput, error) {
	prompt, err := workflow.DecodeValue[string](input.Values["prompt"])
	if err != nil {
		return workflow.ActionOutput{}, err
	}

	return workflow.ActionOutput{Values: map[string]workflow.Value{
		"image": workflow.MustValueOf("preview:" + prompt),
	}}, nil
}

func main() {
	stringSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"string"}`))
	config, _ := json.Marshal(workflow.ActionConfig{Action: "render-preview", Version: "v1"})
	registry, _ := workflow.NewDefaultRegistry(partialRunExampleAction{schema: stringSchema})
	definition := workflow.Definition{
		Schema: workflow.SchemaV1Alpha1, ID: "preview", Revision: "v1", Name: "Preview",
		Inputs: map[string]workflow.WorkflowInput{"prompt": {Schema: stringSchema, Required: true}},
		Outputs: map[string]workflow.OutputBinding{
			"image": {
				Schema: stringSchema,
				Binding: workflow.Binding{
					Source: workflow.BindingNodeOutput, Node: "render", Port: "image",
				},
			},
		},
		Nodes: []workflow.NodeDefinition{
			{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
			{
				ID: "render", Type: workflow.NodeTypeAction, Version: workflow.BuiltinNodeVersion,
				Config: config,
				Inputs: map[string]workflow.Binding{
					"prompt": {
						Source: workflow.BindingNodeOutput, Node: "start", Port: "prompt",
					},
				},
			},
			{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
		},
		Edges: []workflow.ControlEdge{
			{From: workflow.NodeRoute{Node: "start", Route: workflow.RouteSuccess}, To: "render"},
			{From: workflow.NodeRoute{Node: "render", Route: workflow.RouteSuccess}, To: "end"},
		},
		Limits: workflow.DefaultLimits(),
	}
	plan, _ := workflow.Compile(context.Background(), definition, registry)
	runner, _ := workflow.NewRunner()

	first, _ := runner.RunPartial(context.Background(), plan, "render", workflow.PartialRunInput{
		Inputs: map[string]workflow.Value{"prompt": workflow.MustValueOf("paper city")},
	})
	second, _ := runner.RunPartial(context.Background(), plan, "render", workflow.PartialRunInput{
		Inputs:   map[string]workflow.Value{},
		Previous: &first.Data,
	})

	fmt.Println(second.Outputs["image"].String())
	fmt.Println(second.Nodes["start"].Origin, second.Nodes["render"].Origin)

}

Output:
"preview:paper city"
reused executed

type RunnerOption 

type RunnerOption func(*runnerConfig) error

RunnerOption configures a Runner.

func WithCheckpointStore 

func WithCheckpointStore(store CheckpointStore) RunnerOption

WithCheckpointStore configures opaque checkpoint persistence for interrupted Runs and later Resume calls.

func WithClock 

func WithClock(clock func() time.Time) RunnerOption

WithClock replaces the Runner clock, primarily for deterministic tests.

func WithEventSink 

func WithEventSink(sink EventSink) RunnerOption

WithEventSink observes lifecycle metadata without input, output, config, or Action error payloads.

func WithNodeExecutionRecorder 

func WithNodeExecutionRecorder(recorder NodeExecutionRecorder) RunnerOption

WithNodeExecutionRecorder observes sensitive, storage-neutral node execution snapshots from ordinary Run and Resume operations.

Example 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/rsbin1178/pips/workflow"
)

type nodeExecutionExampleAction struct {
	schema workflow.PortSchema
}

func (a nodeExecutionExampleAction) Spec() workflow.ActionSpec {
	return workflow.ActionSpec{
		Key: "record-example", Version: "v1",
		Inputs:  map[string]workflow.PortSchema{},
		Outputs: map[string]workflow.PortSchema{"result": a.schema},
	}
}

func (nodeExecutionExampleAction) Run(
	context.Context,
	workflow.ActionInput,
) (workflow.ActionOutput, error) {
	return workflow.ActionOutput{Values: map[string]workflow.Value{
		"result": workflow.MustValueOf("stored"),
	}}, nil
}

func main() {
	stringSchema, _ := workflow.ParsePortSchema([]byte(`{"type":"string"}`))
	definition := workflow.Definition{
		Schema: workflow.SchemaV1Alpha1, ID: "record-example", Revision: "v1", Name: "Record example",
		Inputs: map[string]workflow.WorkflowInput{},
		Outputs: map[string]workflow.OutputBinding{
			"result": {
				Schema: stringSchema,
				Binding: workflow.Binding{
					Source: workflow.BindingNodeOutput, Node: "action", Port: "result",
				},
			},
		},
		Nodes: []workflow.NodeDefinition{
			{ID: "start", Type: workflow.NodeTypeStart, Version: workflow.BuiltinNodeVersion},
			{
				ID: "action", Type: workflow.NodeTypeAction, Version: workflow.BuiltinNodeVersion,
				Config: json.RawMessage(`{"action":"record-example","version":"v1"}`),
			},
			{ID: "end", Type: workflow.NodeTypeEnd, Version: workflow.BuiltinNodeVersion},
		},
		Edges: []workflow.ControlEdge{
			{From: workflow.NodeRoute{Node: "start", Route: workflow.RouteSuccess}, To: "action"},
			{From: workflow.NodeRoute{Node: "action", Route: workflow.RouteSuccess}, To: "end"},
		},
		Limits: workflow.DefaultLimits(),
	}
	registry, _ := workflow.NewDefaultRegistry(nodeExecutionExampleAction{schema: stringSchema})
	plan, _ := workflow.Compile(context.Background(), definition, registry)

	// The host's recorder can upsert by RunID + Address. It handles its own
	// storage timeout and persistence errors without returning them to Runner.
	runner, _ := workflow.NewRunner(
		workflow.WithRunIDSource(func(time.Time) (string, error) { return "history-run", nil }),
		workflow.WithNodeExecutionRecorder(func(ctx context.Context, record workflow.NodeExecution) {
			storeCtx, cancel := context.WithTimeout(ctx, time.Second)
			defer cancel()

			if record.Address.NodeID != "action" || record.NodeRun.Status == workflow.NodeStatusRunning {
				return
			}

			// A real application would call its repository here and log any error.
			_ = storeCtx

			fmt.Println(record.RunID, record.Address.NodeID, record.NodeRun.Status, record.Route)
		}),
	)
	_, _ = runner.Run(context.Background(), plan, map[string]workflow.Value{})

}

Output:
history-run action succeeded success

func WithRunIDSource 

func WithRunIDSource(source RunIDSource) RunnerOption

WithRunIDSource replaces Run ID generation, primarily for deterministic tests and host correlation policies.

type ScopeFrame 

type ScopeFrame struct {
	Kind   ScopeKind
	NodeID NodeID
	Index  int
}

ScopeFrame identifies the composite node that owns one child execution. Index is -1 for SubWorkflow and the input index for Batch or Loop.

type ScopeKind 

type ScopeKind string

ScopeKind identifies one nested execution boundary. 

const (
	ScopeSubWorkflow   ScopeKind = "sub_workflow"
	ScopeBatchItem     ScopeKind = "batch_item"
	ScopeLoopIteration ScopeKind = "loop_iteration"
)

Nested execution scope kinds.

type SelectorCase 

type SelectorCase struct {
	Route     string    `json:"route"`
	Predicate Predicate `json:"predicate"`
}

SelectorCase maps one structured predicate to a control route.

type SelectorConfig 

type SelectorConfig struct {
	Cases        []SelectorCase `json:"cases"`
	DefaultRoute string         `json:"default_route"`
}

SelectorConfig defines ordered cases and one fallback route.

type SelectorNode 

type SelectorNode struct{}

SelectorNode chooses the first matching route or its default route.

Example 
package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/rsbin1178/pips/workflow"
)

func main() {
	trueValue := workflow.MustValueOf(true)
	config, _ := json.Marshal(workflow.SelectorConfig{
		Cases: []workflow.SelectorCase{
			{
				Route: "approved",
				Predicate: workflow.Predicate{
					Op: workflow.PredicateEqual, Input: "value", Value: &trueValue,
				},
			},
		},
		DefaultRoute: "rejected",
	})
	executor, _ := (workflow.SelectorNode{}).Compile(
		context.Background(),
		nil,
		workflow.NodeDefinition{
			ID:     "selector",
			Config: config,
			Inputs: map[string]workflow.Binding{
				"value": {
					Source: workflow.BindingLiteral,
					Value:  &trueValue,
				},
			},
		},
	)
	output, _ := executor.Invoke(context.Background(), workflow.NodeInput{
		Values: map[string]workflow.Value{"value": trueValue},
	})

	fmt.Println(output.Route)
}

Output:
approved

func (SelectorNode) Compile 

func (SelectorNode) Compile(
	_ context.Context,
	_ CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (SelectorNode) Spec 

func (SelectorNode) Spec() NodeTypeSpec

Spec implements NodeType.

type SetVariableAssignment 

type SetVariableAssignment struct {
	Target string `json:"target"`
	Input  string `json:"input"`
}

SetVariableAssignment maps one node input to one Loop variable.

type SetVariableConfig 

type SetVariableConfig struct {
	Assignments []SetVariableAssignment `json:"assignments"`
}

SetVariableConfig declares an atomic group of Loop-variable assignments.

type SetVariableNode 

type SetVariableNode struct{}

SetVariableNode atomically assigns values to Loop-local variables.

func (SetVariableNode) Compile 

func (SetVariableNode) Compile(
	_ context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (SetVariableNode) Spec 

func (SetVariableNode) Spec() NodeTypeSpec

Spec implements NodeType.

type StartNode 

type StartNode struct{}

StartNode declares Workflow inputs and starts control flow.

func (StartNode) Compile 

func (StartNode) Compile(
	_ context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (StartNode) Spec 

func (StartNode) Spec() NodeTypeSpec

Spec implements NodeType.

type SubWorkflowConfig 

type SubWorkflowConfig struct {
	Workflow DefinitionRef `json:"workflow"`
}

SubWorkflowConfig pins the child Workflow executed by a SubWorkflow node.

type SubWorkflowNode 

type SubWorkflowNode struct{}

SubWorkflowNode synchronously executes one exact referenced Workflow.

func (SubWorkflowNode) Compile 

func (SubWorkflowNode) Compile(
	ctx context.Context,
	compileContext CompileContext,
	definition NodeDefinition,
) (CompiledNode, error)

Compile implements NodeType.

func (SubWorkflowNode) Spec 

func (SubWorkflowNode) Spec() NodeTypeSpec

Spec implements NodeType.

type Value 

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

Value is an immutable, canonical JSON value. Its zero value is invalid; construct values with ParseValue or ValueOf.

func GetInterruptState 

func GetInterruptState(ctx context.Context) (
	wasInterrupted bool,
	hasState bool,
	state Value,
)

GetInterruptState reports state saved by the previously interrupted direct invocation.

func GetResumeContext 

func GetResumeContext(ctx context.Context) (
	isResumeTarget bool,
	hasData bool,
	data Value,
)

GetResumeContext reports whether the current invocation is a direct target or an ancestor of one. Only a direct target receives data.

func MustValueOf 

func MustValueOf(value any) Value

MustValueOf is like ValueOf but panics when value is not JSON-compatible. It is intended for package-level declarations and tests.

func ParseValue 

func ParseValue(data []byte) (Value, error)

ParseValue validates and snapshots one JSON value.

func ValueOf 

func ValueOf(value any) (Value, error)

ValueOf converts a JSON-encodable Go value into an immutable Value.

func (Value) Any 

func (v Value) Any() (any, error)

Any returns an independent standard-library representation of v.

func (Value) Equal 

func (v Value) Equal(other Value) bool

Equal reports equality of the canonical, number-preserving JSON snapshots. Object key order does not affect the result.

func (Value) IsValid 

func (v Value) IsValid() bool

IsValid reports whether v contains a parsed JSON value.

func (Value) Kind 

func (v Value) Kind() ValueKind

Kind reports the JSON variant represented by v.

func (Value) Lookup 

func (v Value) Lookup(path ...string) (Value, bool)

Lookup resolves an object key or array index path without exposing mutable storage. It returns false when any path segment is absent or incompatible.

func (Value) MarshalJSON 

func (v Value) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (Value) RawJSON 

func (v Value) RawJSON() []byte

RawJSON returns an independent copy of v's canonical JSON representation.

func (Value) String 

func (v Value) String() string

String returns v's canonical JSON representation. Invalid values render as an empty string.

func (*Value) UnmarshalJSON 

func (v *Value) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler.

type ValueKind 

type ValueKind uint8

ValueKind identifies one JSON-compatible Value variant. 

const (
	ValueInvalid ValueKind = iota
	ValueNull
	ValueBool
	ValueNumber
	ValueString
	ValueArray
	ValueObject
)

Value kinds. The zero value is invalid rather than JSON null.

type WorkflowInput 

type WorkflowInput struct {
	Schema   PortSchema `json:"schema"`
	Required bool       `json:"required,omitempty"`
	Default  *Value     `json:"default,omitempty"`
}

WorkflowInput declares one Workflow-boundary input contract. A non-null Default applies to a missing optional input and to a supplied empty string, array, or object of the same kind.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.