flow

package
v0.18.0 Latest Latest
Warning

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

 Go to latest Published: Mar 10, 2026 License: MIT Imports: 22 Imported by: 2

README

Flow Package

The Flow package provides execution patterns for the go-command framework, offering flexible ways to run command handlers in different configurations.

Key Features

  • Serial Execution: Run handlers sequentially
  • Parallel Execution: Run handlers concurrently
  • Batch Processing: Process multiple messages in batches
  • Conditional Execution: Execute handlers based on predicates
  • Chain Execution: Sequential execution with message transformation

Installation

go get github.com/goliatone/go-command/flow

FSM v2 Quickstart

The canonical FSM runtime API is envelope-based:

  • ApplyEvent(ctx, ApplyEventRequest[T]) -> *ApplyEventResponse[T]
  • Snapshot(ctx, SnapshotRequest[T]) -> *Snapshot
  • Execute(ctx, msg) is retained only as a compatibility wrapper.
cfg := flow.StateMachineConfig{
  Entity:          "order",
  ExecutionPolicy: flow.ExecutionPolicyLightweight, // or flow.ExecutionPolicyOrchestrated
  States: []flow.StateConfig{
    {Name: "draft", Initial: true},
    {Name: "approved"},
  },
  Transitions: []flow.TransitionConfig{
    {Name: "approve", From: "draft", To: "approved", Action: "audit"},
  },
}

req := flow.TransitionRequest[OrderMsg]{
  StateKey:     func(m OrderMsg) string { return m.ID },
  Event:        func(m OrderMsg) string { return m.Event },
}

actions := flow.NewActionRegistry[OrderMsg]()
_ = actions.Register("audit", func(ctx context.Context, m OrderMsg) error { return nil })

def := cfg.ToMachineDefinition()
store := flow.NewInMemoryStateStore()
_, _ = store.SaveIfVersion(context.Background(), &flow.StateRecord{
  EntityID:       "order-1",
  State:          "draft",
  MachineID:      def.ID,
  MachineVersion: def.Version,
}, 0)
sm, _ := flow.NewStateMachineFromDefinition(
  def,
  store,
  req,
  flow.NewResolverMap[OrderMsg](),
  actions,
  flow.WithExecutionPolicy[OrderMsg](cfg.ExecutionPolicy),
)

res, err := sm.ApplyEvent(context.Background(), flow.ApplyEventRequest[OrderMsg]{
  MachineID: "order",
  EntityID: "order-1",
  Event:    "approve",
  Msg:      OrderMsg{ID: "order-1", Event: "approve"},
  ExecCtx:  flow.ExecutionContext{ActorID: "user-1", Roles: []string{"admin"}, Tenant: "acme"},
  IdempotencyKey: "approve-order-1-v1", // optional
  Metadata: map[string]any{
    "request_id": "req-123",
  },
  DryRun: false, // optional
})
if err != nil {
  // handle runtime category: ErrInvalidTransition / ErrGuardRejected / ErrVersionConflict /
  // ErrIdempotencyConflict / ErrOrchestrationDegraded / ...
}
_ = res.EventID
_ = res.Version
_ = res.Transition
_ = res.Snapshot
_ = res.Execution // nil for lightweight; set for orchestrated policy
_ = res.IdempotencyHit

When DryRun is true, apply remains evaluation-only and skips state/outbox/orchestrator/lifecycle/idempotency-store writes.

Snapshot includes target metadata for static and dynamic transitions:

snap, err := sm.Snapshot(context.Background(), flow.SnapshotRequest[OrderMsg]{
  MachineID: "order",
  EntityID: "order-1",
  Msg:      OrderMsg{ID: "order-1", State: "draft"},
  ExecCtx:  flow.ExecutionContext{ActorID: "user-1"},
  EvaluateGuards: true,
  IncludeBlocked: true,
})
if err != nil {
  // handle error
}
for _, tr := range snap.AllowedTransitions {
  // tr.Allowed indicates whether guards pass for provided msg/execCtx.
  // tr.Rejections contains structured guard rejection details when blocked.
  // tr.Target.Kind => "static" | "dynamic"
  // tr.Target.To / tr.Target.Resolver / tr.Target.Resolved / tr.Target.ResolvedTo / tr.Target.Candidates
}

DSL and UI Schema

def, err := flow.CompileDSL(`
machine onboarding version v2 {
    initial draft
    state draft
    state approved

    transition approve {
        from draft
        to approved
        step order.audit
    }
}`)
if err != nil {
  // dsl validation diagnostics
}

catalog := &flow.EditorCatalog{
  Steps: []flow.CatalogItem{{ID: "order.audit", Label: "order.audit"}},
}
schema, diags := flow.GenerateMachineSchema(def, catalog, nil)
ui := flow.GenerateMachineUISchema(schema)
_ = diags
_ = ui

Orchestrator and RPC Surface

  • Execution policy is mandatory: lightweight or orchestrated.
  • Orchestrated mode exposes execution control:
    • ExecutionStatus(ctx, executionID)
    • PauseExecution(ctx, executionID)
    • ResumeExecution(ctx, executionID)
    • StopExecution(ctx, executionID)
  • Durable orchestrator owns outbox-to-scheduler progression and exposes dispatcher runtime controls:
    • Run(ctx) for continuous claim/enqueue/ack processing
    • RunOnce(ctx) for one managed dispatch cycle
    • StopDispatcher(ctx) for graceful runner shutdown
    • DispatcherStatus() and DispatcherHealth(ctx) for runtime visibility
    • dead-letter inspection via DeadLetters(ctx, DeadLetterScope{...})
    • dispatch outcomes: completed, retry_scheduled, dead_lettered
  • RPC command methods:
    • fsm.apply_event
    • fsm.execution.pause
    • fsm.execution.resume
    • fsm.execution.stop
  • RPC query methods:
    • fsm.snapshot
    • fsm.execution.status
    • fsm.execution.list
    • fsm.execution.history
  • Execution query responses add request telemetry using query_* keys (for example, query_request_id) and keep transition metadata keys immutable.
  • Execution control/status requests support scope fields: machineId, entityId, executionId, tenant.
registry := command.NewRegistry()
server := rpc.NewServer(rpc.WithFailureMode(rpc.FailureModeRecover))
_ = registry.AddResolver("rpc-explicit", rpc.Resolver(server))
_ = flow.RegisterFSMRPCCommands(registry, sm)
_ = registry.Initialize()

Transport helpers map runtime categories to protocol status surfaces:

mapped := flow.MapRuntimeError(err) // HTTP/gRPC/RPC mapping
rpcErr := flow.RPCErrorForError(err)
_ = mapped
_ = rpcErr

See migration guidance: docs/FSM_V2_MIGRATION.md.

Usage Examples

Serial Execution

Run handlers one after another:

// Create handlers
inventoryHandler := &InventoryHandler{}
paymentHandler := &PaymentHandler{}
notificationHandler := &NotificationHandler{}

// Create serial executor with struct handlers
serialExecutor := flow.NewSerialExecutor(
    []command.Commander[ProcessOrderMessage]{
        inventoryHandler,
        paymentHandler,
        notificationHandler,
    },
    runner.WithTimeout(5*time.Second),
)

// Execute
if err := serialExecutor.Execute(ctx, order); err != nil {
    fmt.Printf("Serial execution failed: %v\n", err)
}

// Alternative using functions
err := flow.SerialExecute(
    ctx,
    order,
    []command.CommandFunc[ProcessOrderMessage]{
        logOrderHandler,
        func(ctx context.Context, msg ProcessOrderMessage) error {
            fmt.Printf("Processing order %s inline\n", msg.OrderID)
            return nil
        },
    },
)
Parallel Execution

Run handlers concurrently:

// Create parallel executor
parallelExecutor := flow.NewParallelExecutor(
    []command.Commander[ProcessOrderMessage]{
        inventoryHandler,
        paymentHandler,
        notificationHandler,
    },
    runner.WithTimeout(5*time.Second),
)

if err := parallelExecutor.Execute(ctx, order); err != nil {
    fmt.Printf("Parallel execution failed: %v\n", err)
}
Batch Processing

Process messages in batches with a single handler:

// Create multiple orders
orders := []ProcessOrderMessage{
    {OrderID: "ORD-1001", CustomerID: "C1", Items: []string{"Item1"}, TotalAmount: 10.99},
    {OrderID: "ORD-1002", CustomerID: "C2", Items: []string{"Item2"}, TotalAmount: 20.99},
    // ...more orders
}

// Create batch executor
batchExecutor := flow.NewBatchExecutor(
    logOrderHandler,
    flow.WithBatchSize[ProcessOrderMessage](2),
    flow.WithConcurrency[ProcessOrderMessage](2),
)

if err := batchExecutor.Execute(ctx, orders); err != nil {
    fmt.Printf("Batch execution failed: %v\n", err)
}
Conditional Execution

Execute handlers based on conditions:

// Create conditional executor
conditionalExecutor := flow.NewConditionalExecutor(
    []flow.Conditional[ProcessOrderMessage]{
        {
            Predicate: func(msg ProcessOrderMessage) bool {
                return msg.TotalAmount > 1000.0
            },
            Handler: func(ctx context.Context, msg ProcessOrderMessage) error {
                fmt.Println("High-value order, need manager approval")
                return nil
            },
        },
        {
            Predicate: func(msg ProcessOrderMessage) bool {
                return msg.TotalAmount <= 1000.0
            },
            Handler: func(ctx context.Context, msg ProcessOrderMessage) error {
                fmt.Println("Standard order processing")
                return nil
            },
        },
    },
)

conditionalExecutor.Execute(ctx, order)
Chain Execution

Execute handlers sequentially with message transformation:

// Create chain executor
chainExecutor := flow.NewChainExecutor(
    // Transform message after first handler
    func(msg ProcessOrderMessage) ProcessOrderMessage {
        msg.Status = "INVENTORY_CHECKED"
        return msg
    },
    // Transform message after second handler
    func(msg ProcessOrderMessage) ProcessOrderMessage {
        msg.Status = "PAYMENT_PROCESSED"
        return msg
    },
)

chainExecutor.Execute(ctx, order)
Combined Execution

Combine different execution patterns:

// Parallel inventory check
parallelInventory := flow.NewParallelExecutor(
    []command.Commander[ProcessOrderMessage]{
        &InventoryHandler{Name: "Inventory-US"},
        &InventoryHandler{Name: "Inventory-EU"},
    },
)

// Serial payment and notification
serialNotifyPay := flow.NewSerialExecutor(
    []command.Commander[ProcessOrderMessage]{
        paymentHandler,
        notificationHandler,
    },
)

// Combine them
combinedExecutor := flow.NewSerialExecutor(
    []command.Commander[ProcessOrderMessage]{
        parallelInventory,
        serialNotifyPay,
    },
)

combinedExecutor.Execute(ctx, order)

Configuration Options

Runner Options

All executors accept runner options for configuration:

executor := flow.NewSerialExecutor(
    handlers,
    runner.WithTimeout(5*time.Second),
    runner.WithMaxRetries(3),
    runner.WithExitOnError(true),
    runner.WithLogger(customLogger),
)
Batch Executor Options

Batch executors have additional configuration options:

batchExecutor := flow.NewBatchExecutor(
    handler,
    flow.WithBatchSize[MyMessage](100),   // Process 100 messages per batch
    flow.WithConcurrency[MyMessage](5),   // Run 5 batches concurrently
)

Error Handling

All executors provide consistent error handling:

  • Serial Executor: Returns on first error or aggregates all errors
  • Parallel Executor: Aggregates all errors from concurrent handlers
  • Batch Executor: Aggregates errors from batch processing
  • Conditional Executor: Returns error from matched handler
  • Chain Executor: Returns error from any handler in the chain

Configure error behavior with runner.WithExitOnError:

// Stop on first error
executor := flow.NewSerialExecutor(
    handlers,
    runner.WithExitOnError(true),
)

// Continue on error (aggregate all errors)
executor := flow.NewSerialExecutor(
    handlers,
    runner.WithExitOnError(false),
)

Design Benefits

  • Composable: Mix and match execution patterns
  • Type-safe: Leverages Go generics for type safety
  • Flexible: Configure timeouts, retries, and error handling
  • Consistent: Same error handling and context propagation across patterns
  • Extensible: Easy to add new execution patterns

Configuration (JSON/YAML)

Flows can be defined in config files and built via flow/config_loader.go. Example:

version: 1
flows:
  - id: order_pipeline
    type: serial
    serial:
      steps: ["inventory", "payment", "notify"]
    options:
      timeout: 5s
      max_retries: 2
      exit_on_error: true

Registries (handlers, guards, actions, metrics recorders) resolve the IDs referenced in config. Namespacing helpers avoid ID collisions (namespace::id).

State Machine Usage

Define states, transitions, guards, and actions, with pluggable state stores (in-memory, sqlite, redis). State must be explicitly persisted before transition application.

smCfg := flow.StateMachineConfig{
  Entity:          "order",
  ExecutionPolicy: flow.ExecutionPolicyLightweight,
  States: []flow.StateConfig{{Name: "draft", Initial: true}, {Name: "approved"}},
  Transitions: []flow.TransitionConfig{
    {Name: "approve", From: "draft", To: "approved", Guard: "is_admin", Action: "audit"},
  },
}
guards := flow.NewGuardRegistry[OrderMsg]()
guards.Register("is_admin", func(m OrderMsg) bool { return m.Admin })
actions := flow.NewActionRegistry[OrderMsg]()
actions.Register("audit", func(ctx context.Context, m OrderMsg) error { return nil })
store := flow.NewInMemoryStateStore()
req := flow.TransitionRequest[OrderMsg]{
  StateKey:     func(m OrderMsg) string { return m.ID },
  Event:        func(m OrderMsg) string { return m.Event },
}
def := smCfg.ToMachineDefinition()
resolvers := flow.NewResolverMap[OrderMsg]()
guard, _ := guards.Lookup("is_admin")
resolvers.RegisterGuard("is_admin", guard)
sm, _ := flow.NewStateMachineFromDefinition(
  def,
  store,
  req,
  resolvers,
  actions,
  flow.WithExecutionPolicy[OrderMsg](smCfg.ExecutionPolicy),
)
_, _ = store.SaveIfVersion(ctx, &flow.StateRecord{
  EntityID:       order.ID,
  State:          "draft",
  MachineID:      def.ID,
  MachineVersion: def.Version,
}, 0)
result, err := sm.ApplyEvent(ctx, flow.ApplyEventRequest[OrderMsg]{
  EntityID: order.ID,
  Event:    "approve",
  Msg:      order,
})
if err != nil {
  // handle transition error
}
_ = result

StateMachine also implements Execute(ctx, msg) error for command.Commander[T] compatibility.

Metrics/Tracing Decorators

Use MetricsDecorator to wrap any Flow[T] with metrics; provide a recorder implementation and optionally register it in MetricsRecorderRegistry for config-driven wiring. CircuitBreaker supports half-open probes; RetryableFlow wraps any Flow[T] with a retry strategy.

Hybrid Handler/Mux Usage

Flows accept explicit handlers; a mux resolver adapter is available for registry-driven dispatch. Compose nested flows by converting them to command.Commander[T] via flow.AsCommander. Use the namespacing helpers and registries to prevent ID conflicts across modules.

Registry Resolver Notes

The mux resolver relies on go-command registry metadata. If your command uses an interface message parameter, implement command.MessageFactory to provide a concrete, non-nil message value, otherwise resolver based registration treats the command as unsupported and skips metadata driven integrations.

Legacy Dispatcher Helpers

chain_dispatcher.go and parallel_dispatcher.go are legacy wrappers over the global dispatcher; the recommended approach is to use the flow executors with handler/mux resolvers.

Configuration (JSON/YAML)

Flows can be defined in config files and built via flow/config_loader.go. Example:

version: 1
flows:
  - id: order_pipeline
    type: serial
    serial:
      steps: ["inventory", "payment", "notify"]
    options:
      timeout: 5s
      max_retries: 2
      exit_on_error: true

Registries (handlers, guards, actions, metrics recorders) are used to resolve the IDs referenced in config.

State Machine Usage

Define states, transitions, guards, and actions, with pluggable state stores (in-memory, sqlite, redis):

smCfg := flow.StateMachineConfig{
  Entity:          "order",
  ExecutionPolicy: flow.ExecutionPolicyLightweight,
  States: []flow.StateConfig{{Name: "draft", Initial: true}, {Name: "approved"}},
  Transitions: []flow.TransitionConfig{{Name: "approve", From: "draft", To: "approved", Guard: "is_admin"}},
}
guards := flow.NewGuardRegistry[OrderMsg]()
guards.Register("is_admin", func(m OrderMsg) bool { return m.Admin })
store := flow.NewInMemoryStateStore()
req := flow.TransitionRequest[OrderMsg]{StateKey: func(m OrderMsg) string { return m.ID }, Event: func(m OrderMsg) string { return m.Event }}
def := smCfg.ToMachineDefinition()
resolvers := flow.NewResolverMap[OrderMsg]()
guard, _ := guards.Lookup("is_admin")
resolvers.RegisterGuard("is_admin", guard)
sm, _ := flow.NewStateMachineFromDefinition(def, store, req, resolvers, nil, flow.WithExecutionPolicy[OrderMsg](smCfg.ExecutionPolicy))
_, _ = store.SaveIfVersion(ctx, &flow.StateRecord{
  EntityID:       order.ID,
  State:          "draft",
  MachineID:      def.ID,
  MachineVersion: def.Version,
}, 0)
result, err := sm.ApplyEvent(ctx, flow.ApplyEventRequest[OrderMsg]{
  EntityID: order.ID,
  Event:    "approve",
  Msg:      order,
})
if err != nil {
  // handle transition error
}
_ = result

Metrics/Tracing Decorators

Use MetricsDecorator to wrap any Flow[T] with metrics, provide a recorder implementation and optionally register it in MetricsRecorderRegistry for config-driven wiring.

Hybrid Handler/Mux Usage

Flows accept explicit handlers; a mux resolver adapter is available for registry driven dispatch. Compose nested flows by converting them to command.Commander[T] via flow.AsCommander. Namespacing helpers avoid ID conflicts when registering handlers/guards/actions/recorders.

Documentation

Index

Constants

View Source 
const (
	SeverityError   = "error"
	SeverityWarning = "warning"
	SeverityInfo    = "info"
)
View Source 
const (
	DiagCodeParseError           = "FSM000_PARSE_ERROR"
	DiagCodeInvalidTarget        = "FSM001_INVALID_TARGET"
	DiagCodeUnresolvedAction     = "FSM001_UNRESOLVED_ACTION"
	DiagCodeInvalidWorkflowNode  = "FSM002_INVALID_WORKFLOW_NODE"
	DiagCodeUnknownState         = "FSM003_UNKNOWN_STATE"
	DiagCodeDuplicateTransition  = "FSM004_DUPLICATE_TRANSITION"
	DiagCodeMissingWorkflow      = "FSM005_MISSING_WORKFLOW"
	DiagCodeInvalidGuard         = "FSM006_INVALID_GUARD"
	DiagCodeInvalidDuration      = "FSM007_INVALID_DURATION"
	DiagCodeUnresolvedResolver   = "FSM008_UNRESOLVED_RESOLVER"
	DiagCodeUnresolvedGuard      = "FSM009_UNRESOLVED_GUARD"
	DiagCodeDraftPublishRejected = "FSM010_DRAFT_PUBLISH_REJECTED"
)
View Source 
const (
	FSMRPCMethodApplyEvent       = "fsm.apply_event"
	FSMRPCMethodSnapshot         = "fsm.snapshot"
	FSMRPCMethodExecutionStatus  = "fsm.execution.status"
	FSMRPCMethodExecutionPause   = "fsm.execution.pause"
	FSMRPCMethodExecutionResume  = "fsm.execution.resume"
	FSMRPCMethodExecutionStop    = "fsm.execution.stop"
	FSMRPCMethodExecutionList    = "fsm.execution.list"
	FSMRPCMethodExecutionHistory = "fsm.execution.history"
)
View Source 
const (
	GuardClassificationPass              = "pass"
	GuardClassificationDomainReject      = "domain reject"
	GuardClassificationUnexpectedFailure = "unexpected failure"
)
View Source 
const (
	// LifecycleActivityChannelFSM is the canonical channel for FSM lifecycle activity.
	LifecycleActivityChannelFSM = "fsm"
	// LifecycleActivityObjectTypeMachine is the canonical object type for FSM lifecycle activity.
	LifecycleActivityObjectTypeMachine = "fsm.machine"
	// LifecycleActivityVerbPrefix is the canonical lifecycle verb prefix.
	LifecycleActivityVerbPrefix = "fsm.transition."
)
View Source 
const (
	ExecutionStatePending   = "pending"
	ExecutionStateRunning   = "running"
	ExecutionStatePaused    = "paused"
	ExecutionStateStopped   = "stopped"
	ExecutionStateCompleted = "completed"
	ExecutionStateDegraded  = "degraded"
	ExecutionStateFailed    = "failed"
)
View Source 
const (
	ErrCodeInvalidTransition     = "FSM_INVALID_TRANSITION"
	ErrCodeGuardRejected         = "FSM_GUARD_REJECTED"
	ErrCodeStateNotFound         = "FSM_STATE_NOT_FOUND"
	ErrCodeVersionConflict       = "FSM_VERSION_CONFLICT"
	ErrCodePreconditionFailed    = "FSM_PRECONDITION_FAILED"
	ErrCodeIdempotencyConflict   = "FSM_IDEMPOTENCY_CONFLICT"
	ErrCodeIdempotencyDegraded   = "FSM_IDEMPOTENCY_DEGRADED"
	ErrCodeOrchestrationDegraded = "FSM_ORCHESTRATION_DEGRADED"
)
View Source 
const (
	GRPCCodeAborted            = "Aborted"
	GRPCCodeAlreadyExists      = "AlreadyExists"
	GRPCCodeFailedPrecondition = "FailedPrecondition"
	GRPCCodeInternal           = "Internal"
	GRPCCodeNotFound           = "NotFound"
	GRPCCodePermissionDenied   = "PermissionDenied"
	GRPCCodeUnavailable        = "Unavailable"
)

Variables

View Source 
var (
	ErrInvalidTransition = apperrors.New("invalid transition", apperrors.CategoryBadInput).
							WithTextCode(ErrCodeInvalidTransition)
	ErrGuardRejected = apperrors.New("guard rejected", apperrors.CategoryBadInput).
						WithTextCode(ErrCodeGuardRejected)
	ErrStateNotFound = apperrors.New("state not found", apperrors.CategoryBadInput).
						WithTextCode(ErrCodeStateNotFound)
	ErrVersionConflict = apperrors.New("version conflict", apperrors.CategoryConflict).
						WithTextCode(ErrCodeVersionConflict)
	ErrPreconditionFailed = apperrors.New("precondition failed", apperrors.CategoryBadInput).
							WithTextCode(ErrCodePreconditionFailed)
	ErrIdempotencyConflict = apperrors.New("idempotency conflict", apperrors.CategoryConflict).
							WithTextCode(ErrCodeIdempotencyConflict)
	ErrOrchestrationDegraded = apperrors.New("orchestration degraded", apperrors.CategoryExternal).
								WithTextCode(ErrCodeOrchestrationDegraded)
)
View Source 
var ErrCircuitOpen = fmt.Errorf("circuit open")
View Source 
var ErrIdempotencyRecordExists = errors.New("idempotency record already exists")
View Source 
var (
	// ErrStateVersionConflict indicates optimistic-lock compare-and-set failure.
	ErrStateVersionConflict = errors.New("state version conflict")
)

Functions

func AsCommander added in v0.7.0 

func AsCommander[T any](flow Flow[T]) command.Commander[T]

AsCommander converts a Flow to a command.Commander.

func BuildFlows added in v0.7.0 

func BuildFlows[T command.Message](ctx context.Context, cfg FlowSet, bctx BuildContext[T]) (map[string]Flow[T], error)

BuildFlows constructs flows from config using provided registries.

func ExecuteBatch 

func ExecuteBatch[T command.Message](ctx context.Context, messages []T, handler command.CommandFunc[T], batchSize, concurrency int, opts ...runner.Option) error

ExecuteBatch processes messages in batches with a function handler

func GRPCCodeForError added in v0.15.0 

func GRPCCodeForError(err error) string

GRPCCodeForError returns the mapped gRPC status code string for an engine error.

func GenerateMachineSchema added in v0.15.0 

func GenerateMachineSchema(def *MachineDefinition, catalog *EditorCatalog, scope *ValidationScope) (*MachineSchema, []ValidationDiagnostic)

GenerateMachineSchema performs MachineDefinition -> MachineSchema conversion.

func HTTPStatusForError added in v0.15.0 

func HTTPStatusForError(err error) int

HTTPStatusForError returns the mapped HTTP status code for an engine error.

func IsNonRetryable added in v0.15.0 

func IsNonRetryable(err error) bool

IsNonRetryable reports whether err is marked terminal for retries.

func MachineDefinitionToUISchema added in v0.15.0 

func MachineDefinitionToUISchema(def *MachineDefinition, catalog *EditorCatalog, scope *ValidationScope) (*MachineUISchema, []ValidationDiagnostic)

MachineDefinitionToUISchema is the full MachineDefinition -> MachineSchema -> MachineUISchema pipeline.

func MarshalFlowSet added in v0.7.0 

func MarshalFlowSet(cfg FlowSet) ([]byte, error)

MarshalFlowSet renders FlowSet as JSON (useful for fixtures).

func NewFSMRPCCommands added in v0.15.0 

func NewFSMRPCCommands[T command.Message](machine *StateMachine[T]) []any

NewFSMRPCCommands returns the full FSM RPC method family as registry commands.

func ParallelExecute 

func ParallelExecute[T any](ctx context.Context, msg T, handlers []command.CommandFunc[T], opts ...runner.Option) error

ParallelExecute runs handlers concurrently with function handlers

func RegisterFSMRPCCommands added in v0.15.0 

func RegisterFSMRPCCommands[T command.Message](registry *command.Registry, machine *StateMachine[T]) error

RegisterFSMRPCCommands registers the full FSM method family into a command registry.

func SerialExecute added in v0.7.0 

func SerialExecute[T any](ctx context.Context, msg T, handlers []command.CommandFunc[T], opts ...runner.Option) error

SerialExecute will run each handler in sequence with function handlers

func SetDSLCompileKnownActions added in v0.15.0 

func SetDSLCompileKnownActions(actionIDs []string)

SetDSLCompileKnownActions configures known action IDs used for unresolved-action validation.

Types

type ActionRegistry added in v0.7.0 

type ActionRegistry[T any] struct {
	// contains filtered or unexported fields
}

ActionRegistry stores named actions executed during transitions.

func NewActionRegistry added in v0.7.0 

func NewActionRegistry[T any]() *ActionRegistry[T]

NewActionRegistry creates an empty registry.

func (*ActionRegistry[T]) IDs added in v0.15.0 

func (r *ActionRegistry[T]) IDs() []string

IDs returns sorted action IDs for deterministic catalog generation.

func (*ActionRegistry[T]) Lookup added in v0.7.0 

func (r *ActionRegistry[T]) Lookup(name string) (func(context.Context, T) error, bool)

Lookup retrieves an action by name.

func (*ActionRegistry[T]) Register added in v0.7.0 

func (r *ActionRegistry[T]) Register(name string, action func(context.Context, T) error) error

Register adds an action by name.

func (*ActionRegistry[T]) RegisterNamespaced added in v0.7.0 

func (r *ActionRegistry[T]) RegisterNamespaced(namespace, name string, action func(context.Context, T) error) error

RegisterNamespaced adds an action under namespace+name.

func (*ActionRegistry[T]) SetNamespacer added in v0.7.0 

func (r *ActionRegistry[T]) SetNamespacer(fn func(string, string) string)

SetNamespacer customizes how action IDs are namespaced.

type AggregateErrorStrategy added in v0.7.0 

type AggregateErrorStrategy struct{}

AggregateErrorStrategy combines all errors into one

func (AggregateErrorStrategy) HandleErrors added in v0.7.0 

func (a AggregateErrorStrategy) HandleErrors(errs []error) error

type ApplyEventRequest added in v0.15.0 

type ApplyEventRequest[T any] struct {
	MachineID       string
	EntityID        string
	Event           string
	Msg             T
	ExecCtx         ExecutionContext
	ExpectedState   string
	ExpectedVersion int
	IdempotencyKey  string
	Metadata        map[string]any
	DryRun          bool
}

ApplyEventRequest is the canonical runtime envelope for transitions.

type ApplyEventResponse added in v0.15.0 

type ApplyEventResponse[T any] struct {
	EventID        string
	Version        int
	Transition     *TransitionResult[T]
	Snapshot       *Snapshot
	Execution      *ExecutionHandle
	IdempotencyHit bool
}

ApplyEventResponse is the canonical transport-agnostic transition envelope.

type BatchConfig added in v0.7.0 

type BatchConfig struct {
	Handler     string      `json:"handler" yaml:"handler"`
	BatchSize   int         `json:"batch_size,omitempty" yaml:"batch_size,omitempty"`
	Concurrency int         `json:"concurrency,omitempty" yaml:"concurrency,omitempty"`
	Opts        FlowOptions `json:"options,omitempty" yaml:"options,omitempty"`
}

type BatchExecutor 

type BatchExecutor[T command.Message] struct {
	// contains filtered or unexported fields
}

BatchExecutor processes commands in batches

func NewBatchExecutor 

func NewBatchExecutor[T command.Message](handler command.Commander[T], opts ...BatchExecutorOption[T]) *BatchExecutor[T]

func (*BatchExecutor[T]) Execute 

func (b *BatchExecutor[T]) Execute(ctx context.Context, messages []T) error

type BatchExecutorOption 

type BatchExecutorOption[T command.Message] func(*BatchExecutor[T])

func WithBatchSize 

func WithBatchSize[T command.Message](size int) BatchExecutorOption[T]

func WithConcurrency 

func WithConcurrency[T command.Message](n int) BatchExecutorOption[T]

func WithRunnerOptions added in v0.7.0 

func WithRunnerOptions[T command.Message](opts ...Option) BatchExecutorOption[T]

WithRunnerOptions attaches runner options to the batch executor.

type BuildContext added in v0.7.0 

type BuildContext[T command.Message] struct {
	Handlers *HandlerRegistry[T]
	Guards   *GuardRegistry[T]
	Actions  *ActionRegistry[T]
	Store    StateStore
	Request  TransitionRequest[T]
}

BuildContext bundles registries and stores needed to construct flows from config.

type CatalogItem added in v0.15.0 

type CatalogItem struct {
	ID          string                 `json:"id"`
	Label       string                 `json:"label"`
	Category    string                 `json:"category"`
	Schema      map[string]any         `json:"schema,omitempty"`
	UI          UIComponent            `json:"ui"`
	Description string                 `json:"description,omitempty"`
	Metadata    map[string]interface{} `json:"metadata,omitempty"`
}

CatalogItem is one palette entry backed by runtime metadata.

type CircuitBreaker added in v0.7.0 

type CircuitBreaker[T command.Message] struct {
	// contains filtered or unexported fields
}

CircuitBreaker implements the circuit breaker pattern

func NewCircuitBreaker added in v0.7.0 

func NewCircuitBreaker[T command.Message](
	flow Flow[T],
	failureThreshold int,
	resetTimeout time.Duration,
	opts ...CircuitBreakerOption[T],
) *CircuitBreaker[T]

func (*CircuitBreaker[T]) Execute added in v0.7.0 

func (c *CircuitBreaker[T]) Execute(ctx context.Context, msg T) error

type CircuitBreakerOption added in v0.7.0 

type CircuitBreakerOption[T command.Message] func(*CircuitBreaker[T])

CircuitBreakerOption allows customizing breaker behavior.

func WithHalfOpenProbe added in v0.7.0 

func WithHalfOpenProbe[T command.Message](probe Flow[T]) CircuitBreakerOption[T]

WithHalfOpenProbe sets the flow executed when probing a half-open circuit.

type ClaimedOutboxEntry added in v0.15.0 

type ClaimedOutboxEntry struct {
	OutboxEntry
	LeaseToken string
}

ClaimedOutboxEntry carries a claimed entry and proof-of-ownership lease token.

type CommandEffect added in v0.15.0 

type CommandEffect struct {
	ActionID string
	Payload  map[string]any
	Async    bool
	Delay    time.Duration
	Timeout  time.Duration
	Metadata map[string]any
}

CommandEffect models command-backed side effects.

type CompiledMachine added in v0.15.0 

type CompiledMachine[T any] struct {
	ID          string
	Version     string
	States      []State
	Transitions []CompiledTransition[T]
}

CompiledMachine is the executable machine contract.

func CompileMachine added in v0.15.0 

func CompileMachine[T any](def *MachineDefinition, reg ResolverRegistry[T]) (*CompiledMachine[T], error)

CompileMachine compiles canonical authoring definitions into runtime contracts.

type CompiledTransition added in v0.15.0 

type CompiledTransition[T any] struct {
	ID              string
	Event           string
	From            string
	To              string
	DynamicTo       DynamicTargetResolver[T]
	DynamicResolver string
	Guards          []Guard[T]
	Plan            CompiledWorkflowPlan
	Metadata        map[string]any
}

CompiledTransition is the executable transition contract.

type CompiledWorkflowNode added in v0.15.0 

type CompiledWorkflowNode struct {
	ID            string
	Kind          string
	Step          *Step
	ConditionExpr string
	Next          []string
	Metadata      map[string]any
}

CompiledWorkflowNode is a compiled workflow graph node.

type CompiledWorkflowPlan added in v0.15.0 

type CompiledWorkflowPlan struct {
	Nodes []CompiledWorkflowNode
}

CompiledWorkflowPlan is the compiled transition workflow graph.

type Conditional added in v0.7.0 

type Conditional[T command.Message] struct {
	// Predicate is evaluated when provided.
	Predicate func(T) bool
	// Guard references a named guard in the registry when Predicate is nil.
	Guard   string
	Handler func(context.Context, T) error
}

type ConditionalBranch added in v0.7.0 

type ConditionalBranch struct {
	Guard   string `json:"guard" yaml:"guard"`
	Handler string `json:"handler" yaml:"handler"`
}

type ConditionalConfig added in v0.7.0 

type ConditionalConfig struct {
	Branches       []ConditionalBranch `json:"branches" yaml:"branches"`
	DefaultHandler string              `json:"default_handler,omitempty" yaml:"default_handler,omitempty"`
}

type ConditionalExecutor added in v0.7.0 

type ConditionalExecutor[T command.Message] struct {
	// contains filtered or unexported fields
}

ConditionalExecutor enables conditional command execution based on predicates or guards.

func NewConditionalExecutor added in v0.7.0 

func NewConditionalExecutor[T command.Message](branches []Conditional[T], opts ...ConditionalOption[T]) *ConditionalExecutor[T]

func (*ConditionalExecutor[T]) Execute added in v0.7.0 

func (b *ConditionalExecutor[T]) Execute(ctx context.Context, msg T) error

type ConditionalOption added in v0.7.0 

type ConditionalOption[T command.Message] func(*ConditionalExecutor[T])

ConditionalOption customizes conditional executors.

func WithDefaultHandler added in v0.7.0 

func WithDefaultHandler[T command.Message](handler func(context.Context, T) error) ConditionalOption[T]

WithDefaultHandler sets the handler executed when no branch matches.

func WithGuardRegistry added in v0.7.0 

func WithGuardRegistry[T command.Message](registry *GuardRegistry[T]) ConditionalOption[T]

WithGuardRegistry wires a guard registry for guard-based branches.

type DSLCompileOptions added in v0.15.0 

type DSLCompileOptions struct {
	KnownActions map[string]struct{}
}

DSLCompileOptions configures compile-time validation behavior.

type DSLValidationError added in v0.15.0 

type DSLValidationError struct {
	Diagnostics []ValidationDiagnostic
}

DSLValidationError wraps deterministic diagnostics emitted by DSL compilation/validation.

func (*DSLValidationError) Error added in v0.15.0 

func (e *DSLValidationError) Error() string

type DeadLetterScope added in v0.15.0 

type DeadLetterScope struct {
	MachineID   string
	EntityID    string
	ExecutionID string
	Limit       int
}

DeadLetterScope constrains dead-letter inspection queries.

type DecoratorConfig added in v0.7.0 

type DecoratorConfig struct {
	Type   string         `json:"type" yaml:"type"`
	Config map[string]any `json:"config,omitempty" yaml:"config,omitempty"`
}

type DispatchEntryResult added in v0.15.0 

type DispatchEntryResult struct {
	OutboxID     string
	ExecutionID  string
	EntityID     string
	TransitionID string
	Event        string
	Attempt      int
	Outcome      DispatchOutcome
	RetryAt      time.Time
	Error        string
	OccurredAt   time.Time
	Metadata     map[string]any
}

DispatchEntryResult captures one outbox entry dispatch result.

type DispatchOutcome added in v0.15.0 

type DispatchOutcome string

DispatchOutcome classifies one dispatch attempt result. 

const (
	DispatchOutcomeCompleted      DispatchOutcome = "completed"
	DispatchOutcomeRetryScheduled DispatchOutcome = "retry_scheduled"
	DispatchOutcomeDeadLettered   DispatchOutcome = "dead_lettered"
)

type DispatchReport added in v0.15.0 

type DispatchReport struct {
	WorkerID   string
	Claimed    int
	Processed  int
	Lag        time.Duration
	StartedAt  time.Time
	FinishedAt time.Time
	Outcomes   []DispatchEntryResult
}

DispatchReport summarizes one dispatcher cycle.

type DispatchRetryOwner added in v0.15.0 

type DispatchRetryOwner string

DispatchRetryOwner defines who owns retry/backoff decisions. 

const (
	DispatchRetryOwnerDispatcher DispatchRetryOwner = "dispatcher"
	DispatchRetryOwnerExternal   DispatchRetryOwner = "external"
)

type DispatcherHealth added in v0.15.0 

type DispatcherHealth struct {
	Healthy bool
	Reason  string
	Status  DispatcherRuntimeStatus
}

DispatcherHealth reports health derived from runtime status.

type DispatcherMetrics added in v0.15.0 

type DispatcherMetrics interface {
	RecordDispatchLag(duration time.Duration)
	RecordDispatchOutcome(outcome DispatchOutcome)
	RecordRetryAttempt(attempt int)
	RecordOrchestrationDegraded(reason string)
}

DispatcherMetrics captures observability events for dispatch runtime behavior.

type DispatcherRuntime added in v0.15.0 

type DispatcherRuntime interface {
	Run(ctx context.Context) error
	RunOnce(ctx context.Context) (DispatchReport, error)
	Stop(ctx context.Context) error
	Status() DispatcherRuntimeStatus
	Health(ctx context.Context) DispatcherHealth
}

DispatcherRuntime exposes managed dispatcher execution controls.

type DispatcherRuntimeState added in v0.15.0 

type DispatcherRuntimeState string

DispatcherRuntimeState tracks lifecycle of the background dispatch runner. 

const (
	DispatcherRuntimeStateIdle     DispatcherRuntimeState = "idle"
	DispatcherRuntimeStateRunning  DispatcherRuntimeState = "running"
	DispatcherRuntimeStateStopping DispatcherRuntimeState = "stopping"
	DispatcherRuntimeStateStopped  DispatcherRuntimeState = "stopped"
)

type DispatcherRuntimeStatus added in v0.15.0 

type DispatcherRuntimeStatus struct {
	WorkerID            string
	State               DispatcherRuntimeState
	LastRunAt           time.Time
	LastSuccessAt       time.Time
	LastError           string
	ConsecutiveFailures int
	LastClaimed         int
	LastProcessed       int
	LastLag             time.Duration
}

DispatcherRuntimeStatus captures the latest runtime state and cycle metrics.

type DraftMachineDocument added in v0.15.0 

type DraftMachineDocument struct {
	Definition *MachineDefinition `json:"definition"`
	UISchema   *MachineUISchema   `json:"ui_schema"`
	DraftState DraftState         `json:"draft_state"`
}

DraftMachineDocument persists editor drafts that may be incomplete/invalid.

func (*DraftMachineDocument) CanPublish added in v0.15.0 

func (d *DraftMachineDocument) CanPublish(catalog *EditorCatalog) (bool, []ValidationDiagnostic)

CanPublish validates and reports whether draft can be published.

func (*DraftMachineDocument) Publish added in v0.15.0 

func (d *DraftMachineDocument) Publish(catalog *EditorCatalog) error

Publish marks the draft as publishable only when no error diagnostics remain.

func (*DraftMachineDocument) Validate added in v0.15.0 

func (d *DraftMachineDocument) Validate(catalog *EditorCatalog, scope *ValidationScope) []ValidationDiagnostic

Validate returns draft diagnostics for current definition.

type DraftState added in v0.15.0 

type DraftState struct {
	IsDraft     bool      `json:"is_draft"`
	LastSavedAt time.Time `json:"last_saved_at"`
}

type DurableOrchestrator added in v0.15.0 

type DurableOrchestrator[T command.Message] struct {
	// contains filtered or unexported fields
}

DurableOrchestrator persists execution metadata and lifecycle intents.

func NewDurableOrchestrator added in v0.15.0 

func NewDurableOrchestrator[T command.Message](
	records ExecutionRecordStore[T],
	scheduler JobScheduler,
	outbox OutboxStore,
	opts ...DurableOrchestratorOption[T],
) (*DurableOrchestrator[T], error)

NewDurableOrchestrator builds a durable orchestrator implementation.

func (*DurableOrchestrator[T]) DeadLetters added in v0.15.0 

func (o *DurableOrchestrator[T]) DeadLetters(ctx context.Context, scope DeadLetterScope) ([]OutboxEntry, error)

DeadLetters exposes dead-letter inspection through the orchestrator's outbox store.

func (*DurableOrchestrator[T]) DispatchHistory added in v0.15.0 

func (o *DurableOrchestrator[T]) DispatchHistory(ctx context.Context, scope ExecutionScope) ([]ExecutionDispatchHistory, error)

DispatchHistory returns durable dispatch progression records for inspection.

func (*DurableOrchestrator[T]) DispatchRuntime added in v0.15.0 

func (o *DurableOrchestrator[T]) DispatchRuntime() DispatcherRuntime

DispatchRuntime exposes the orchestrator-managed dispatcher runtime.

func (*DurableOrchestrator[T]) DispatcherHealth added in v0.15.0 

func (o *DurableOrchestrator[T]) DispatcherHealth(ctx context.Context) (DispatcherHealth, error)

DispatcherHealth returns dispatcher runtime health and emits health hooks.

func (*DurableOrchestrator[T]) DispatcherStatus added in v0.15.0 

func (o *DurableOrchestrator[T]) DispatcherStatus() (DispatcherRuntimeStatus, error)

DispatcherStatus returns the latest dispatcher runtime status snapshot.

func (*DurableOrchestrator[T]) HandleResume added in v0.15.0 

func (o *DurableOrchestrator[T]) HandleResume(ctx context.Context, req ResumeRequest[T]) error

HandleResume validates expected state/version and classifies stale paths as terminal.

func (*DurableOrchestrator[T]) History added in v0.15.0 

func (o *DurableOrchestrator[T]) History(ctx context.Context, scope ExecutionScope) ([]TransitionLifecycleEvent[T], error)

func (*DurableOrchestrator[T]) LifecycleIntents added in v0.15.0 

func (o *DurableOrchestrator[T]) LifecycleIntents(ctx context.Context) ([]TransitionLifecycleEvent[T], error)

LifecycleIntents returns stored lifecycle intents (testing/inspection helper).

func (*DurableOrchestrator[T]) List added in v0.15.0 

func (o *DurableOrchestrator[T]) List(ctx context.Context, scope ExecutionScope) ([]ExecutionStatus, error)

func (*DurableOrchestrator[T]) OnTransitionLifecycleEvent added in v0.15.0 

func (o *DurableOrchestrator[T]) OnTransitionLifecycleEvent(ctx context.Context, evt TransitionLifecycleEvent[T]) error

func (*DurableOrchestrator[T]) Pause added in v0.15.0 

func (o *DurableOrchestrator[T]) Pause(ctx context.Context, executionID string) error

func (*DurableOrchestrator[T]) Resume added in v0.15.0 

func (o *DurableOrchestrator[T]) Resume(ctx context.Context, executionID string) error

func (*DurableOrchestrator[T]) Run added in v0.15.0 

func (o *DurableOrchestrator[T]) Run(ctx context.Context) error

Run starts the orchestrator-managed dispatcher runtime loop.

func (*DurableOrchestrator[T]) RunOnce added in v0.15.0 

func (o *DurableOrchestrator[T]) RunOnce(ctx context.Context) (DispatchReport, error)

RunOnce executes one orchestrator-managed dispatch cycle.

func (*DurableOrchestrator[T]) Start added in v0.15.0 

func (o *DurableOrchestrator[T]) Start(ctx context.Context, req StartRequest[T]) (*ExecutionHandle, error)

func (*DurableOrchestrator[T]) Status added in v0.15.0 

func (o *DurableOrchestrator[T]) Status(ctx context.Context, executionID string) (*ExecutionStatus, error)

func (*DurableOrchestrator[T]) Stop added in v0.15.0 

func (o *DurableOrchestrator[T]) Stop(ctx context.Context, executionID string) error

func (*DurableOrchestrator[T]) StopDispatcher added in v0.15.0 

func (o *DurableOrchestrator[T]) StopDispatcher(ctx context.Context) error

StopDispatcher stops the orchestrator-managed dispatcher runtime loop.

type DurableOrchestratorOption added in v0.15.0 

type DurableOrchestratorOption[T command.Message] func(*DurableOrchestrator[T])

DurableOrchestratorOption customizes durable orchestration.

func WithDurableDispatchAutoRun added in v0.15.0 

func WithDurableDispatchAutoRun[T command.Message](enabled bool) DurableOrchestratorOption[T]

WithDurableDispatchAutoRun configures automatic RunOnce progression during Start.

func WithDurableDispatcherRunner added in v0.15.0 

func WithDurableDispatcherRunner[T command.Message](runner DispatcherRuntime) DurableOrchestratorOption[T]

WithDurableDispatcherRunner overrides the default orchestrator-managed dispatcher runtime.

func WithDurableLifecycleIntentStore added in v0.15.0 

func WithDurableLifecycleIntentStore[T command.Message](store LifecycleIntentStore[T]) DurableOrchestratorOption[T]

WithDurableLifecycleIntentStore configures lifecycle intent persistence.

func WithDurableLogger added in v0.15.0 

func WithDurableLogger[T command.Message](logger Logger) DurableOrchestratorOption[T]

WithDurableLogger configures durable orchestrator logging.

func WithDurableOutboxDispatcherOptions added in v0.15.0 

func WithDurableOutboxDispatcherOptions[T command.Message](opts ...OutboxDispatcherOption) DurableOrchestratorOption[T]

WithDurableOutboxDispatcherOptions configures options for the default managed outbox dispatcher runtime.

func WithDurableRetryPolicy added in v0.15.0 

func WithDurableRetryPolicy[T command.Message](policy RetryPolicy) DurableOrchestratorOption[T]

WithDurableRetryPolicy configures retry ownership defaults.

type DynamicTargetDefinition added in v0.15.0 

type DynamicTargetDefinition struct {
	Resolver string
}

DynamicTargetDefinition declares a target resolver reference.

type DynamicTargetResolver added in v0.15.0 

type DynamicTargetResolver[T any] func(ctx context.Context, msg T, execCtx ExecutionContext) (string, error)

DynamicTargetResolver resolves transition targets at runtime.

type EdgeLayout added in v0.15.0 

type EdgeLayout struct {
	SourcePort string  `json:"source_port,omitempty"`
	TargetPort string  `json:"target_port,omitempty"`
	Points     []Point `json:"points,omitempty"`
}

type EditorCatalog added in v0.15.0 

type EditorCatalog struct {
	Guards    []CatalogItem `json:"guards"`
	Steps     []CatalogItem `json:"steps"`
	Resolvers []CatalogItem `json:"resolvers"`
}

EditorCatalog represents palette/introspection data for authoring surfaces.

func BuildEditorCatalog added in v0.15.0 

func BuildEditorCatalog[T any](guards *GuardRegistry[T], actions *ActionRegistry[T], resolvers *ResolverMap[T]) EditorCatalog

BuildEditorCatalog derives palette entries from runtime registries.

type Effect added in v0.15.0 

type Effect interface{}

Effect models an orchestration side effect emitted by transitions.

type EmitEvent added in v0.15.0 

type EmitEvent struct {
	Event    string
	Msg      any
	Metadata map[string]any
}

EmitEvent models runtime-emitted follow-up events.

type ErrorStrategy added in v0.7.0 

type ErrorStrategy interface {
	HandleErrors([]error) error
}

ErrorStrategy defines how to handle multiple errors from parallel execution

type ExecutionContext added in v0.15.0 

type ExecutionContext struct {
	ActorID string
	Roles   []string
	Tenant  string
}

ExecutionContext carries caller identity and tenancy information.

type ExecutionDispatchHistory added in v0.15.0 

type ExecutionDispatchHistory struct {
	ExecutionID  string
	MachineID    string
	EntityID     string
	TransitionID string
	OutboxID     string
	Event        string
	Outcome      DispatchOutcome
	Attempt      int
	RetryAt      time.Time
	Error        string
	OccurredAt   time.Time
	Metadata     map[string]any
}

ExecutionDispatchHistory captures durable dispatch progression for one execution.

type ExecutionHandle added in v0.15.0 

type ExecutionHandle struct {
	ExecutionID string
	Policy      string
	Status      string
	Metadata    map[string]any
}

ExecutionHandle describes external orchestration execution state.

type ExecutionHistoryProvider added in v0.15.0 

type ExecutionHistoryProvider[T any] interface {
	History(ctx context.Context, scope ExecutionScope) ([]TransitionLifecycleEvent[T], error)
}

ExecutionHistoryProvider exposes execution lifecycle history for query handlers.

type ExecutionListProvider added in v0.15.0 

type ExecutionListProvider interface {
	List(ctx context.Context, scope ExecutionScope) ([]ExecutionStatus, error)
}

ExecutionListProvider exposes execution listing for query handlers.

type ExecutionMessage added in v0.15.0 

type ExecutionMessage struct {
	ID        string
	Topic     string
	Payload   []byte
	Metadata  map[string]any
	CreatedAt time.Time
}

ExecutionMessage is the scheduler payload boundary used by orchestrators.

type ExecutionPolicy added in v0.15.0 

type ExecutionPolicy string

ExecutionPolicy controls how transition effects and lifecycle events are handled. 

const (
	ExecutionPolicyLightweight  ExecutionPolicy = "lightweight"
	ExecutionPolicyOrchestrated ExecutionPolicy = "orchestrated"
)

type ExecutionRecord added in v0.15.0 

type ExecutionRecord[T any] struct {
	ExecutionID     string
	Policy          ExecutionPolicy
	Status          string
	MachineID       string
	MachineVersion  string
	EntityID        string
	Event           string
	TransitionID    string
	PreviousState   string
	CurrentState    string
	ExpectedState   string
	ExpectedVersion int
	AttemptCount    int
	ErrorCode       string
	ErrorMessage    string
	RetryPolicy     RetryPolicy
	Effects         []Effect
	Metadata        map[string]any
	CreatedAt       time.Time
	UpdatedAt       time.Time
	Msg             T
}

ExecutionRecord persists durable execution tracking metadata.

type ExecutionRecordStore added in v0.15.0 

type ExecutionRecordStore[T any] interface {
	Save(ctx context.Context, rec *ExecutionRecord[T]) error
	Load(ctx context.Context, executionID string) (*ExecutionRecord[T], error)
	List(ctx context.Context) ([]*ExecutionRecord[T], error)
	ListByScope(ctx context.Context, scope ExecutionScope) ([]*ExecutionRecord[T], error)
	UpdateStatus(ctx context.Context, executionID, status string) error
	UpdateResult(ctx context.Context, executionID, status, errorCode, errorMessage, currentState string) error
	ApplyDispatchOutcome(ctx context.Context, result DispatchEntryResult) error
	DispatchHistory(ctx context.Context, scope ExecutionScope) ([]ExecutionDispatchHistory, error)
}

ExecutionRecordStore persists execution records for durable orchestration.

type ExecutionScope added in v0.15.0 

type ExecutionScope struct {
	MachineID   string
	EntityID    string
	ExecutionID string
	Tenant      string
}

ExecutionScope constrains execution control/query APIs.

type ExecutionStatus added in v0.15.0 

type ExecutionStatus struct {
	ExecutionID string
	Policy      ExecutionPolicy
	Status      string
	Attempts    int
	ErrorCode   string
	Error       string
	UpdatedAt   time.Time
	Metadata    map[string]any
}

ExecutionStatus reports orchestration progress.

type FSMApplyEventRPCCommand added in v0.15.0 

type FSMApplyEventRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMApplyEventRPCCommand provides the fsm.apply_event method.

func NewFSMApplyEventRPCCommand added in v0.15.0 

func NewFSMApplyEventRPCCommand[T command.Message](machine *StateMachine[T]) *FSMApplyEventRPCCommand[T]

func (*FSMApplyEventRPCCommand[T]) Query added in v0.15.0 

func (c *FSMApplyEventRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMApplyEventRequest[T]],
) (cmdrpc.ResponseEnvelope[*ApplyEventResponse[T]], error)

func (*FSMApplyEventRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMApplyEventRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMApplyEventRequest added in v0.15.0 

type FSMApplyEventRequest[T command.Message] struct {
	MachineID       string         `json:"machineId,omitempty"`
	EntityID        string         `json:"entityId"`
	Event           string         `json:"event"`
	Msg             T              `json:"msg"`
	ExpectedState   string         `json:"expectedState,omitempty"`
	ExpectedVersion int            `json:"expectedVersion,omitempty"`
	IdempotencyKey  string         `json:"idempotencyKey,omitempty"`
	Metadata        map[string]any `json:"metadata,omitempty"`
	DryRun          bool           `json:"dryRun,omitempty"`
}

FSMApplyEventRequest is the RPC request data for fsm.apply_event.

type FSMExecutionControlRequest added in v0.15.0 

type FSMExecutionControlRequest struct {
	MachineID   string `json:"machineId,omitempty"`
	EntityID    string `json:"entityId,omitempty"`
	ExecutionID string `json:"executionId"`
	Tenant      string `json:"tenant,omitempty"`
}

FSMExecutionControlRequest is the RPC request data for execution control/status methods.

type FSMExecutionHistoryRPCCommand added in v0.15.0 

type FSMExecutionHistoryRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMExecutionHistoryRPCCommand provides the fsm.execution.history method.

func NewFSMExecutionHistoryRPCCommand added in v0.15.0 

func NewFSMExecutionHistoryRPCCommand[T command.Message](machine *StateMachine[T]) *FSMExecutionHistoryRPCCommand[T]

func (*FSMExecutionHistoryRPCCommand[T]) Query added in v0.15.0 

func (c *FSMExecutionHistoryRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMExecutionHistoryRequest],
) (cmdrpc.ResponseEnvelope[FSMExecutionHistoryResponse[T]], error)

func (*FSMExecutionHistoryRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMExecutionHistoryRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMExecutionHistoryRequest added in v0.15.0 

type FSMExecutionHistoryRequest struct {
	MachineID   string `json:"machineId,omitempty"`
	EntityID    string `json:"entityId,omitempty"`
	ExecutionID string `json:"executionId,omitempty"`
	Tenant      string `json:"tenant,omitempty"`
}

FSMExecutionHistoryRequest is the RPC request data for execution history queries.

type FSMExecutionHistoryResponse added in v0.15.0 

type FSMExecutionHistoryResponse[T command.Message] struct {
	Items []TransitionLifecycleEvent[T] `json:"items"`
}

FSMExecutionHistoryResponse is the response payload for fsm.execution.history.

type FSMExecutionListRPCCommand added in v0.15.0 

type FSMExecutionListRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMExecutionListRPCCommand provides the fsm.execution.list method.

func NewFSMExecutionListRPCCommand added in v0.15.0 

func NewFSMExecutionListRPCCommand[T command.Message](machine *StateMachine[T]) *FSMExecutionListRPCCommand[T]

func (*FSMExecutionListRPCCommand[T]) Query added in v0.15.0 

func (c *FSMExecutionListRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMExecutionListRequest],
) (cmdrpc.ResponseEnvelope[FSMExecutionListResponse], error)

func (*FSMExecutionListRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMExecutionListRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMExecutionListRequest added in v0.15.0 

type FSMExecutionListRequest struct {
	MachineID   string `json:"machineId,omitempty"`
	EntityID    string `json:"entityId,omitempty"`
	ExecutionID string `json:"executionId,omitempty"`
	Tenant      string `json:"tenant,omitempty"`
}

FSMExecutionListRequest is the RPC request data for execution list queries.

type FSMExecutionListResponse added in v0.15.0 

type FSMExecutionListResponse struct {
	Items []ExecutionStatus `json:"items"`
}

FSMExecutionListResponse is the response payload for fsm.execution.list.

type FSMExecutionPauseRPCCommand added in v0.15.0 

type FSMExecutionPauseRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMExecutionPauseRPCCommand provides the fsm.execution.pause method.

func NewFSMExecutionPauseRPCCommand added in v0.15.0 

func NewFSMExecutionPauseRPCCommand[T command.Message](machine *StateMachine[T]) *FSMExecutionPauseRPCCommand[T]

func (*FSMExecutionPauseRPCCommand[T]) Query added in v0.15.0 

func (c *FSMExecutionPauseRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMExecutionControlRequest],
) (cmdrpc.ResponseEnvelope[*ExecutionStatus], error)

func (*FSMExecutionPauseRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMExecutionPauseRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMExecutionResumeRPCCommand added in v0.15.0 

type FSMExecutionResumeRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMExecutionResumeRPCCommand provides the fsm.execution.resume method.

func NewFSMExecutionResumeRPCCommand added in v0.15.0 

func NewFSMExecutionResumeRPCCommand[T command.Message](machine *StateMachine[T]) *FSMExecutionResumeRPCCommand[T]

func (*FSMExecutionResumeRPCCommand[T]) Query added in v0.15.0 

func (c *FSMExecutionResumeRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMExecutionControlRequest],
) (cmdrpc.ResponseEnvelope[*ExecutionStatus], error)

func (*FSMExecutionResumeRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMExecutionResumeRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMExecutionScope added in v0.15.0 

type FSMExecutionScope struct {
	MachineID   string `json:"machineId,omitempty"`
	EntityID    string `json:"entityId,omitempty"`
	ExecutionID string `json:"executionId,omitempty"`
	Tenant      string `json:"tenant,omitempty"`
}

FSMExecutionScope constrains execution query/control handlers.

type FSMExecutionStatusRPCCommand added in v0.15.0 

type FSMExecutionStatusRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMExecutionStatusRPCCommand provides the fsm.execution.status method.

func NewFSMExecutionStatusRPCCommand added in v0.15.0 

func NewFSMExecutionStatusRPCCommand[T command.Message](machine *StateMachine[T]) *FSMExecutionStatusRPCCommand[T]

func (*FSMExecutionStatusRPCCommand[T]) Query added in v0.15.0 

func (c *FSMExecutionStatusRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMExecutionControlRequest],
) (cmdrpc.ResponseEnvelope[*ExecutionStatus], error)

func (*FSMExecutionStatusRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMExecutionStatusRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMExecutionStopRPCCommand added in v0.15.0 

type FSMExecutionStopRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMExecutionStopRPCCommand provides the fsm.execution.stop method.

func NewFSMExecutionStopRPCCommand added in v0.15.0 

func NewFSMExecutionStopRPCCommand[T command.Message](machine *StateMachine[T]) *FSMExecutionStopRPCCommand[T]

func (*FSMExecutionStopRPCCommand[T]) Query added in v0.15.0 

func (c *FSMExecutionStopRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMExecutionControlRequest],
) (cmdrpc.ResponseEnvelope[*ExecutionStatus], error)

func (*FSMExecutionStopRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMExecutionStopRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMSnapshotRPCCommand added in v0.15.0 

type FSMSnapshotRPCCommand[T command.Message] struct {
	Machine *StateMachine[T]
	Spec    cmdrpc.EndpointSpec
}

FSMSnapshotRPCCommand provides the fsm.snapshot method.

func NewFSMSnapshotRPCCommand added in v0.15.0 

func NewFSMSnapshotRPCCommand[T command.Message](machine *StateMachine[T]) *FSMSnapshotRPCCommand[T]

func (*FSMSnapshotRPCCommand[T]) Query added in v0.15.0 

func (c *FSMSnapshotRPCCommand[T]) Query(
	ctx context.Context,
	req cmdrpc.RequestEnvelope[FSMSnapshotRequest[T]],
) (cmdrpc.ResponseEnvelope[*Snapshot], error)

func (*FSMSnapshotRPCCommand[T]) RPCEndpoints added in v0.15.0 

func (c *FSMSnapshotRPCCommand[T]) RPCEndpoints() []cmdrpc.EndpointDefinition

type FSMSnapshotRequest added in v0.15.0 

type FSMSnapshotRequest[T command.Message] struct {
	MachineID      string `json:"machineId,omitempty"`
	EntityID       string `json:"entityId"`
	Msg            T      `json:"msg"`
	EvaluateGuards bool   `json:"evaluateGuards,omitempty"`
	IncludeBlocked bool   `json:"includeBlocked,omitempty"`
}

FSMSnapshotRequest is the RPC request data for fsm.snapshot.

type FailFastStrategy added in v0.7.0 

type FailFastStrategy struct{}

FailFastStrategy returns the first error encountered

func (FailFastStrategy) HandleErrors added in v0.7.0 

func (f FailFastStrategy) HandleErrors(errs []error) error

type FieldsLogger added in v0.15.0 

type FieldsLogger interface {
	WithFields(map[string]any) Logger
}

FieldsLogger extends Logger with structured-field support.

type Flow added in v0.7.0 

type Flow[T any] interface {
	Execute(ctx context.Context, msg T) error
}

Flow is the common contract for all flow executors.

type FlowCommander added in v0.7.0 

type FlowCommander[T any] struct {
	// contains filtered or unexported fields
}

FlowCommander wraps a Flow so it can be registered as a command.Commander.

func (*FlowCommander[T]) Execute added in v0.7.0 

func (f *FlowCommander[T]) Execute(ctx context.Context, msg T) error

Execute delegates to the underlying flow.

type FlowDefinition added in v0.7.0 

type FlowDefinition struct {
	ID           string              `json:"id" yaml:"id"`
	Type         string              `json:"type" yaml:"type"`
	Options      FlowOptions         `json:"options,omitempty" yaml:"options,omitempty"`
	Serial       *SerialConfig       `json:"serial,omitempty" yaml:"serial,omitempty"`
	Parallel     *ParallelConfig     `json:"parallel,omitempty" yaml:"parallel,omitempty"`
	Batch        *BatchConfig        `json:"batch,omitempty" yaml:"batch,omitempty"`
	Conditional  *ConditionalConfig  `json:"conditional,omitempty" yaml:"conditional,omitempty"`
	Saga         *SagaConfig         `json:"saga,omitempty" yaml:"saga,omitempty"`
	StateMachine *StateMachineConfig `json:"state_machine,omitempty" yaml:"state_machine,omitempty"`
	Decorators   []DecoratorConfig   `json:"decorators,omitempty" yaml:"decorators,omitempty"`
}

FlowDefinition describes a single flow instance.

func (FlowDefinition) Validate added in v0.7.0 

func (d FlowDefinition) Validate() error

Validate checks required fields for the flow definition.

type FlowOptions added in v0.7.0 

type FlowOptions struct {
	Timeout     time.Duration `json:"timeout,omitempty" yaml:"timeout,omitempty"`
	NoTimeout   bool          `json:"no_timeout,omitempty" yaml:"no_timeout,omitempty"`
	MaxRetries  int           `json:"max_retries,omitempty" yaml:"max_retries,omitempty"`
	MaxRuns     int           `json:"max_runs,omitempty" yaml:"max_runs,omitempty"`
	RunOnce     bool          `json:"run_once,omitempty" yaml:"run_once,omitempty"`
	ExitOnError bool          `json:"exit_on_error,omitempty" yaml:"exit_on_error,omitempty"`
	Deadline    time.Time     `json:"deadline,omitempty" yaml:"deadline,omitempty"`
}

FlowOptions captures common runner options.

type FlowSet added in v0.7.0 

type FlowSet struct {
	Version int              `json:"version" yaml:"version"`
	Flows   []FlowDefinition `json:"flows" yaml:"flows"`
	Options FlowOptions      `json:"options,omitempty" yaml:"options,omitempty"`
	Meta    map[string]any   `json:"meta,omitempty" yaml:"meta,omitempty"`
}

FlowSet represents a collection of flows loaded from config.

func ParseFlowSet added in v0.7.0 

func ParseFlowSet(data []byte) (FlowSet, error)

ParseFlowSet attempts to parse JSON or YAML into a FlowSet.

func (FlowSet) Validate added in v0.7.0 

func (c FlowSet) Validate() error

Validate performs basic structural validation.

type FmtLogger added in v0.15.0 

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

FmtLogger is the local fallback logger used when no external logger is configured.

func NewFmtLogger added in v0.15.0 

func NewFmtLogger(out io.Writer) *FmtLogger

NewFmtLogger constructs a fallback logger writing to stdout when out is nil.

func (*FmtLogger) Debug added in v0.15.0 

func (l *FmtLogger) Debug(msg string, args ...any)

func (*FmtLogger) Error added in v0.15.0 

func (l *FmtLogger) Error(msg string, args ...any)

func (*FmtLogger) Fatal added in v0.15.0 

func (l *FmtLogger) Fatal(msg string, args ...any)

func (*FmtLogger) Info added in v0.15.0 

func (l *FmtLogger) Info(msg string, args ...any)

func (*FmtLogger) Trace added in v0.15.0 

func (l *FmtLogger) Trace(msg string, args ...any)

func (*FmtLogger) Warn added in v0.15.0 

func (l *FmtLogger) Warn(msg string, args ...any)

func (*FmtLogger) WithContext added in v0.15.0 

func (l *FmtLogger) WithContext(ctx context.Context) Logger

func (*FmtLogger) WithFields added in v0.15.0 

func (l *FmtLogger) WithFields(fields map[string]any) Logger

WithFields adds fields on a shallow-copy logger.

type GraphLayout added in v0.15.0 

type GraphLayout struct {
	Viewport Viewport                   `json:"viewport"`
	Nodes    map[string]NodeLayout      `json:"nodes"`
	Edges    map[string]EdgeLayout      `json:"edges"`
	Groups   []GroupLayout              `json:"groups,omitempty"`
	Unknown  map[string]json.RawMessage `json:"-"`
}

GraphLayout persists visual editor geometry and unknown fields for forward compatibility.

func (GraphLayout) MarshalJSON added in v0.15.0 

func (g GraphLayout) MarshalJSON() ([]byte, error)

func (*GraphLayout) UnmarshalJSON added in v0.15.0 

func (g *GraphLayout) UnmarshalJSON(data []byte) error

type GroupLayout added in v0.15.0 

type GroupLayout struct {
	ID      string   `json:"id"`
	Label   string   `json:"label,omitempty"`
	NodeIDs []string `json:"node_ids,omitempty"`
}

type Guard added in v0.15.0 

type Guard[T any] func(ctx context.Context, msg T, execCtx ExecutionContext) error

Guard is a runtime guard predicate.

type GuardDefinition added in v0.15.0 

type GuardDefinition struct {
	Type     string
	Expr     string
	Ref      string
	Metadata map[string]any
}

GuardDefinition describes declarative guard references.

type GuardRegistry added in v0.7.0 

type GuardRegistry[T any] struct {
	// contains filtered or unexported fields
}

GuardRegistry stores named guard functions.

func NewGuardRegistry added in v0.7.0 

func NewGuardRegistry[T any]() *GuardRegistry[T]

NewGuardRegistry creates an empty registry.

func (*GuardRegistry[T]) IDs added in v0.15.0 

func (g *GuardRegistry[T]) IDs() []string

IDs returns sorted guard IDs for deterministic catalog generation.

func (*GuardRegistry[T]) Lookup added in v0.7.0 

func (g *GuardRegistry[T]) Lookup(name string) (Guard[T], bool)

Lookup retrieves a guard by name.

func (*GuardRegistry[T]) Register added in v0.7.0 

func (g *GuardRegistry[T]) Register(name string, guard func(T) bool) error

Register stores a guard by name.

func (*GuardRegistry[T]) RegisterNamespaced added in v0.7.0 

func (g *GuardRegistry[T]) RegisterNamespaced(namespace, name string, guard func(T) bool) error

RegisterNamespaced stores a guard using namespace+name.

func (*GuardRegistry[T]) RegisterWithContext added in v0.15.0 

func (g *GuardRegistry[T]) RegisterWithContext(name string, guard Guard[T]) error

RegisterWithContext stores a context-aware guard by name.

func (*GuardRegistry[T]) RegisterWithContextNamespaced added in v0.15.0 

func (g *GuardRegistry[T]) RegisterWithContextNamespaced(namespace, name string, guard Guard[T]) error

RegisterWithContextNamespaced stores a context-aware guard using namespace+name.

func (*GuardRegistry[T]) SetNamespacer added in v0.7.0 

func (g *GuardRegistry[T]) SetNamespacer(fn func(string, string) string)

SetNamespacer customizes how guard IDs are namespaced.

type GuardRejection added in v0.15.0 

type GuardRejection struct {
	Code            string
	Category        string
	Retryable       bool
	RequiresAction  bool
	Message         string
	RemediationHint string
	Metadata        map[string]any
}

GuardRejection captures structured guard rejection diagnostics.

func (*GuardRejection) Error added in v0.15.0 

func (g *GuardRejection) Error() string

type GuardUISchema added in v0.15.0 

type GuardUISchema struct {
	Type       string         `json:"type"`
	Properties map[string]any `json:"properties,omitempty"`
	UI         UIComponent    `json:"ui"`
}

type HandlerRegistry added in v0.7.0 

type HandlerRegistry[T any] struct {
	// contains filtered or unexported fields
}

HandlerRegistry stores named commanders.

func NewHandlerRegistry added in v0.7.0 

func NewHandlerRegistry[T any]() *HandlerRegistry[T]

NewHandlerRegistry creates an empty registry.

func (*HandlerRegistry[T]) Lookup added in v0.7.0 

func (r *HandlerRegistry[T]) Lookup(id string) (command.Commander[T], bool)

Lookup returns a commander by id.

func (*HandlerRegistry[T]) Register added in v0.7.0 

func (r *HandlerRegistry[T]) Register(id string, h command.Commander[T]) error

Register stores a commander by id.

func (*HandlerRegistry[T]) RegisterNamespaced added in v0.7.0 

func (r *HandlerRegistry[T]) RegisterNamespaced(namespace, id string, h command.Commander[T]) error

RegisterNamespaced stores a commander using a namespace + id.

func (*HandlerRegistry[T]) SetNamespacer added in v0.7.0 

func (r *HandlerRegistry[T]) SetNamespacer(fn func(string, string) string)

SetNamespacer customizes how IDs are namespaced.

type HandlerResolver added in v0.7.0 

type HandlerResolver[T any] struct {
	// contains filtered or unexported fields
}

HandlerResolver resolves a static list of handlers.

func NewHandlerResolver added in v0.7.0 

func NewHandlerResolver[T any](handlers ...command.Commander[T]) *HandlerResolver[T]

NewHandlerResolver constructs a resolver backed by explicit handlers.

func (*HandlerResolver[T]) Resolve added in v0.7.0 

func (r *HandlerResolver[T]) Resolve(_ context.Context, msg T) ([]command.Commander[T], error)

Resolve returns the configured handlers or an error if none exist.

type HookFailureMode added in v0.15.0 

type HookFailureMode string

HookFailureMode controls lifecycle-hook error behavior. 

const (
	HookFailureModeFailOpen   HookFailureMode = "fail_open"
	HookFailureModeFailClosed HookFailureMode = "fail_closed"
)

type IdempotencyRecord added in v0.15.0 

type IdempotencyRecord[T any] struct {
	Scope       IdempotencyScope
	RequestHash string
	Response    *ApplyEventResponse[T]
	CreatedAt   time.Time
}

IdempotencyRecord stores payload fingerprint and response replay data.

type IdempotencyScope added in v0.15.0 

type IdempotencyScope struct {
	MachineID      string
	EntityID       string
	Event          string
	IdempotencyKey string
}

IdempotencyScope identifies one idempotency record boundary.

type IdempotencyStore added in v0.15.0 

type IdempotencyStore[T any] interface {
	Load(ctx context.Context, scope IdempotencyScope) (*IdempotencyRecord[T], error)
	Save(ctx context.Context, rec *IdempotencyRecord[T]) error
}

IdempotencyStore persists idempotency records keyed by machine+entity+event+idempotency-key.

type InMemoryExecutionRecordStore added in v0.15.0 

type InMemoryExecutionRecordStore[T any] struct {
	// contains filtered or unexported fields
}

InMemoryExecutionRecordStore keeps execution records in memory.

func NewInMemoryExecutionRecordStore added in v0.15.0 

func NewInMemoryExecutionRecordStore[T any]() *InMemoryExecutionRecordStore[T]

NewInMemoryExecutionRecordStore constructs an empty execution record store.

func (*InMemoryExecutionRecordStore[T]) ApplyDispatchOutcome added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) ApplyDispatchOutcome(_ context.Context, result DispatchEntryResult) error

func (*InMemoryExecutionRecordStore[T]) DispatchHistory added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) DispatchHistory(_ context.Context, scope ExecutionScope) ([]ExecutionDispatchHistory, error)

func (*InMemoryExecutionRecordStore[T]) List added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) List(_ context.Context) ([]*ExecutionRecord[T], error)

func (*InMemoryExecutionRecordStore[T]) ListByScope added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) ListByScope(_ context.Context, scope ExecutionScope) ([]*ExecutionRecord[T], error)

func (*InMemoryExecutionRecordStore[T]) Load added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) Load(_ context.Context, executionID string) (*ExecutionRecord[T], error)

func (*InMemoryExecutionRecordStore[T]) Save added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) Save(_ context.Context, rec *ExecutionRecord[T]) error

func (*InMemoryExecutionRecordStore[T]) UpdateResult added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) UpdateResult(
	_ context.Context,
	executionID, status, errorCode, errorMessage, currentState string,
) error

func (*InMemoryExecutionRecordStore[T]) UpdateStatus added in v0.15.0 

func (s *InMemoryExecutionRecordStore[T]) UpdateStatus(_ context.Context, executionID, status string) error

type InMemoryIdempotencyStore added in v0.15.0 

type InMemoryIdempotencyStore[T any] struct {
	// contains filtered or unexported fields
}

InMemoryIdempotencyStore keeps idempotency records in memory.

func NewInMemoryIdempotencyStore added in v0.15.0 

func NewInMemoryIdempotencyStore[T any]() *InMemoryIdempotencyStore[T]

NewInMemoryIdempotencyStore constructs an empty in-memory idempotency store.

func (*InMemoryIdempotencyStore[T]) Load added in v0.15.0 

func (s *InMemoryIdempotencyStore[T]) Load(_ context.Context, scope IdempotencyScope) (*IdempotencyRecord[T], error)

func (*InMemoryIdempotencyStore[T]) Save added in v0.15.0 

func (s *InMemoryIdempotencyStore[T]) Save(_ context.Context, rec *IdempotencyRecord[T]) error

type InMemoryJobScheduler added in v0.15.0 

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

InMemoryJobScheduler stores enqueued messages for inspection/testing.

func NewInMemoryJobScheduler added in v0.15.0 

func NewInMemoryJobScheduler() *InMemoryJobScheduler

NewInMemoryJobScheduler constructs an empty in-memory scheduler.

func (*InMemoryJobScheduler) Enqueue added in v0.15.0 

func (s *InMemoryJobScheduler) Enqueue(ctx context.Context, msg *ExecutionMessage) error

func (*InMemoryJobScheduler) EnqueueAfter added in v0.15.0 

func (s *InMemoryJobScheduler) EnqueueAfter(ctx context.Context, msg *ExecutionMessage, delay time.Duration) error

func (*InMemoryJobScheduler) EnqueueAt added in v0.15.0 

func (s *InMemoryJobScheduler) EnqueueAt(ctx context.Context, msg *ExecutionMessage, at time.Time) error

func (*InMemoryJobScheduler) Messages added in v0.15.0 

func (s *InMemoryJobScheduler) Messages() []ScheduledExecutionMessage

Messages returns a copy of all enqueued messages.

func (*InMemoryJobScheduler) SetEnqueueHook added in v0.15.0 

func (s *InMemoryJobScheduler) SetEnqueueHook(fn func(*ExecutionMessage) error)

SetEnqueueHook sets an optional callback used by enqueue operations.

type InMemoryLifecycleIntentStore added in v0.15.0 

type InMemoryLifecycleIntentStore[T any] struct {
	// contains filtered or unexported fields
}

InMemoryLifecycleIntentStore keeps lifecycle intents in memory.

func NewInMemoryLifecycleIntentStore added in v0.15.0 

func NewInMemoryLifecycleIntentStore[T any]() *InMemoryLifecycleIntentStore[T]

NewInMemoryLifecycleIntentStore constructs an empty lifecycle intent store.

func (*InMemoryLifecycleIntentStore[T]) Append added in v0.15.0 

func (s *InMemoryLifecycleIntentStore[T]) Append(_ context.Context, evt TransitionLifecycleEvent[T]) error

func (*InMemoryLifecycleIntentStore[T]) List added in v0.15.0 

func (s *InMemoryLifecycleIntentStore[T]) List(_ context.Context) ([]TransitionLifecycleEvent[T], error)

type InMemoryStateStore added in v0.7.0 

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

InMemoryStateStore is a thread-safe in-memory state and outbox store.

func NewInMemoryStateStore added in v0.7.0 

func NewInMemoryStateStore() *InMemoryStateStore

NewInMemoryStateStore constructs an empty store.

func (*InMemoryStateStore) ClaimPending added in v0.15.0 

func (s *InMemoryStateStore) ClaimPending(
	_ context.Context,
	workerID string,
	limit int,
	leaseTTL time.Duration,
) ([]ClaimedOutboxEntry, error)

ClaimPending claims pending entries with a lease for the given worker.

func (*InMemoryStateStore) ExtendLease added in v0.15.0 

func (s *InMemoryStateStore) ExtendLease(_ context.Context, id, leaseToken string, leaseTTL time.Duration) error

ExtendLease extends a claimed entry lease when token ownership matches.

func (*InMemoryStateStore) ListDeadLetters added in v0.15.0 

func (s *InMemoryStateStore) ListDeadLetters(_ context.Context, scope DeadLetterScope) ([]OutboxEntry, error)

ListDeadLetters returns dead-lettered outbox entries, filtered by optional scope fields.

func (*InMemoryStateStore) Load added in v0.7.0 

func (s *InMemoryStateStore) Load(_ context.Context, key string) (*StateRecord, error)

Load returns a cloned state record for the entity.

func (*InMemoryStateStore) MarkCompleted added in v0.15.0 

func (s *InMemoryStateStore) MarkCompleted(_ context.Context, id, leaseToken string) error

MarkCompleted marks an outbox entry as fully dispatched.

func (*InMemoryStateStore) MarkDeadLetter added in v0.15.0 

func (s *InMemoryStateStore) MarkDeadLetter(_ context.Context, id, leaseToken, reason string) error

MarkDeadLetter marks an outbox entry as terminal and inspectable in dead-letter queries.

func (*InMemoryStateStore) MarkFailed added in v0.15.0 

func (s *InMemoryStateStore) MarkFailed(_ context.Context, id, leaseToken string, retryAt time.Time, reason string) error

MarkFailed marks an outbox entry as failed and schedules retry.

func (*InMemoryStateStore) OutboxEntries added in v0.15.0 

func (s *InMemoryStateStore) OutboxEntries() []OutboxEntry

OutboxEntries returns a cloned outbox slice for assertions and debugging.

func (*InMemoryStateStore) RunInTransaction added in v0.15.0 

func (s *InMemoryStateStore) RunInTransaction(ctx context.Context, fn func(TxStore) error) error

RunInTransaction applies mutations atomically with rollback on error.

func (*InMemoryStateStore) SaveIfVersion added in v0.15.0 

func (s *InMemoryStateStore) SaveIfVersion(_ context.Context, rec *StateRecord, expectedVersion int) (int, error)

SaveIfVersion performs compare-and-set persistence for state records.

type InspectorSchema added in v0.15.0 

type InspectorSchema struct {
	Sections []InspectorSection `json:"sections,omitempty"`
}

type InspectorSection added in v0.15.0 

type InspectorSection struct {
	ID     string   `json:"id"`
	Label  string   `json:"label"`
	Fields []string `json:"fields,omitempty"`
}

type JobScheduler added in v0.15.0 

type JobScheduler interface {
	Enqueue(ctx context.Context, msg *ExecutionMessage) error
	EnqueueAt(ctx context.Context, msg *ExecutionMessage, at time.Time) error
	EnqueueAfter(ctx context.Context, msg *ExecutionMessage, delay time.Duration) error
}

JobScheduler enqueues execution messages for durable workers.

type LifecycleActivityEnvelope added in v0.15.0 

type LifecycleActivityEnvelope struct {
	Channel    string
	Verb       string
	ObjectType string
	ObjectID   string
	ActorID    string
	TenantID   string
	OccurredAt time.Time
	Metadata   map[string]any
}

LifecycleActivityEnvelope is a transport-neutral activity payload projected from lifecycle events.

func BuildLifecycleActivityEnvelope added in v0.15.0 

func BuildLifecycleActivityEnvelope[T any](evt TransitionLifecycleEvent[T]) LifecycleActivityEnvelope

BuildLifecycleActivityEnvelope maps a transition lifecycle event into the canonical activity envelope.

func (LifecycleActivityEnvelope) ActivityActorID added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityActorID() string

func (LifecycleActivityEnvelope) ActivityChannel added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityChannel() string

func (LifecycleActivityEnvelope) ActivityMetadata added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityMetadata() map[string]any

func (LifecycleActivityEnvelope) ActivityObjectID added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityObjectID() string

func (LifecycleActivityEnvelope) ActivityObjectType added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityObjectType() string

func (LifecycleActivityEnvelope) ActivityOccurredAt added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityOccurredAt() time.Time

func (LifecycleActivityEnvelope) ActivityTenantID added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityTenantID() string

func (LifecycleActivityEnvelope) ActivityVerb added in v0.15.0 

func (e LifecycleActivityEnvelope) ActivityVerb() string

type LifecycleActivityHook added in v0.15.0 

type LifecycleActivityHook[T any] struct {
	Sink LifecycleActivitySink
}

LifecycleActivityHook projects lifecycle events into canonical activity envelopes.

func (*LifecycleActivityHook[T]) Notify added in v0.15.0 

func (h *LifecycleActivityHook[T]) Notify(ctx context.Context, evt TransitionLifecycleEvent[T]) error

Notify satisfies TransitionLifecycleHook and forwards projected envelopes to the sink.

type LifecycleActivitySink added in v0.15.0 

type LifecycleActivitySink interface {
	LogLifecycleActivity(ctx context.Context, envelope LifecycleActivityEnvelope) error
}

LifecycleActivitySink receives lifecycle activity envelopes.

type LifecycleActivitySinkFunc added in v0.15.0 

type LifecycleActivitySinkFunc func(ctx context.Context, envelope LifecycleActivityEnvelope) error

LifecycleActivitySinkFunc adapts a function into LifecycleActivitySink.

func (LifecycleActivitySinkFunc) LogLifecycleActivity added in v0.15.0 

func (f LifecycleActivitySinkFunc) LogLifecycleActivity(ctx context.Context, envelope LifecycleActivityEnvelope) error

LogLifecycleActivity satisfies LifecycleActivitySink.

type LifecycleEventHandler added in v0.15.0 

type LifecycleEventHandler[T any] interface {
	OnTransitionLifecycleEvent(ctx context.Context, evt TransitionLifecycleEvent[T]) error
}

LifecycleEventHandler allows orchestrators to own lifecycle handling behavior.

type LifecycleIntentStore added in v0.15.0 

type LifecycleIntentStore[T any] interface {
	Append(ctx context.Context, evt TransitionLifecycleEvent[T]) error
	List(ctx context.Context) ([]TransitionLifecycleEvent[T], error)
}

LifecycleIntentStore persists lifecycle intents for async/durable processing.

type LightweightOrchestrator added in v0.15.0 

type LightweightOrchestrator[T command.Message] struct {
	// contains filtered or unexported fields
}

LightweightOrchestrator executes transition effects in-process.

func NewLightweightOrchestrator added in v0.15.0 

func NewLightweightOrchestrator[T command.Message](
	actions *ActionRegistry[T],
	opts ...LightweightOrchestratorOption[T],
) *LightweightOrchestrator[T]

NewLightweightOrchestrator builds an in-process orchestrator.

func (*LightweightOrchestrator[T]) History added in v0.15.0 

func (o *LightweightOrchestrator[T]) History(_ context.Context, scope ExecutionScope) ([]TransitionLifecycleEvent[T], error)

func (*LightweightOrchestrator[T]) List added in v0.15.0 

func (o *LightweightOrchestrator[T]) List(_ context.Context, scope ExecutionScope) ([]ExecutionStatus, error)

func (*LightweightOrchestrator[T]) OnTransitionLifecycleEvent added in v0.15.0 

func (o *LightweightOrchestrator[T]) OnTransitionLifecycleEvent(ctx context.Context, evt TransitionLifecycleEvent[T]) error

func (*LightweightOrchestrator[T]) Pause added in v0.15.0 

func (o *LightweightOrchestrator[T]) Pause(_ context.Context, executionID string) error

func (*LightweightOrchestrator[T]) Resume added in v0.15.0 

func (o *LightweightOrchestrator[T]) Resume(_ context.Context, executionID string) error

func (*LightweightOrchestrator[T]) Start added in v0.15.0 

func (o *LightweightOrchestrator[T]) Start(ctx context.Context, req StartRequest[T]) (*ExecutionHandle, error)

func (*LightweightOrchestrator[T]) Status added in v0.15.0 

func (o *LightweightOrchestrator[T]) Status(_ context.Context, executionID string) (*ExecutionStatus, error)

func (*LightweightOrchestrator[T]) Stop added in v0.15.0 

func (o *LightweightOrchestrator[T]) Stop(_ context.Context, executionID string) error

type LightweightOrchestratorOption added in v0.15.0 

type LightweightOrchestratorOption[T command.Message] func(*LightweightOrchestrator[T])

LightweightOrchestratorOption customizes lightweight orchestration.

func WithLightweightHookFailureMode added in v0.15.0 

func WithLightweightHookFailureMode[T command.Message](mode HookFailureMode) LightweightOrchestratorOption[T]

WithLightweightHookFailureMode configures hook failure handling.

func WithLightweightHooks added in v0.15.0 

func WithLightweightHooks[T command.Message](hooks ...TransitionLifecycleHook[T]) LightweightOrchestratorOption[T]

WithLightweightHooks sets lifecycle hooks for in-process fan-out.

func WithLightweightLogger added in v0.15.0 

func WithLightweightLogger[T command.Message](logger Logger) LightweightOrchestratorOption[T]

WithLightweightLogger configures orchestration logs.

type Logger added in v0.15.0 

type Logger interface {
	Trace(msg string, args ...any)
	Debug(msg string, args ...any)
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
	Fatal(msg string, args ...any)
	WithContext(ctx context.Context) Logger
}

Logger is the runtime logging contract.

type MachineDefinition added in v0.15.0 

type MachineDefinition struct {
	ID          string
	Name        string
	Version     string
	States      []StateDefinition
	Transitions []TransitionDefinition
}

MachineDefinition is the canonical authoring/interchange contract.

func CompileDSL added in v0.15.0 

func CompileDSL(input string) (*MachineDefinition, error)

CompileDSL parses and validates DSL into canonical MachineDefinition.

func CompileDSLWithOptions added in v0.15.0 

func CompileDSLWithOptions(input string, opts DSLCompileOptions) (*MachineDefinition, error)

CompileDSLWithOptions parses DSL into canonical MachineDefinition with explicit compile options.

func MachineDefinitionFromUISchema added in v0.15.0 

func MachineDefinitionFromUISchema(ui *MachineUISchema, id, version string) *MachineDefinition

MachineDefinitionFromUISchema projects editor graph data back into canonical machine definitions.

func NormalizeMachineDefinition added in v0.15.0 

func NormalizeMachineDefinition(def *MachineDefinition) *MachineDefinition

NormalizeMachineDefinition canonicalizes ordering, IDs, and duration fields for stable import/export.

func (*MachineDefinition) ToDSL added in v0.15.0 

func (m *MachineDefinition) ToDSL() (string, error)

ToDSL renders canonical machine definitions to deterministic DSL.

type MachineSchema added in v0.15.0 

type MachineSchema struct {
	ID          string                 `json:"id"`
	Name        string                 `json:"name"`
	Version     string                 `json:"version"`
	States      []StateDefinition      `json:"states"`
	Transitions []TransitionDefinition `json:"transitions"`
	Catalog     EditorCatalog          `json:"catalog,omitempty"`
	Diagnostics []ValidationDiagnostic `json:"diagnostics,omitempty"`
}

MachineSchema is the canonical intermediate schema used before UI projection.

type MachineUISchema added in v0.15.0 

type MachineUISchema struct {
	Layout    string             `json:"layout"`
	Nodes     []StateNodeSchema  `json:"nodes"`
	Edges     []TransitionSchema `json:"edges"`
	Inspector InspectorSchema    `json:"inspector"`
	Graph     GraphLayout        `json:"graph_layout,omitempty"`
}

MachineUISchema contains editor-facing graph and inspector representations.

func GenerateMachineUISchema added in v0.15.0 

func GenerateMachineUISchema(schema *MachineSchema) *MachineUISchema

GenerateMachineUISchema performs MachineSchema -> MachineUISchema conversion.

type MetricsDecorator added in v0.7.0 

type MetricsDecorator[T command.Message] struct {
	// contains filtered or unexported fields
}

MetricsDecorator adds metrics to any flow

func NewMetricsDecorator added in v0.7.0 

func NewMetricsDecorator[T command.Message](
	flow interface {
		Execute(context.Context, T) error
	},
	recorder MetricsRecorder,
) *MetricsDecorator[T]

func (*MetricsDecorator[T]) Execute added in v0.7.0 

func (m *MetricsDecorator[T]) Execute(ctx context.Context, msg T) error

type MetricsRecorder added in v0.7.0 

type MetricsRecorder interface {
	RecordDuration(name string, duration time.Duration)
	RecordError(name string)
	RecordSuccess(name string)
}

type MetricsRecorderRegistry added in v0.7.0 

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

MetricsRecorderRegistry stores named metrics recorders.

func NewMetricsRecorderRegistry added in v0.7.0 

func NewMetricsRecorderRegistry() *MetricsRecorderRegistry

NewMetricsRecorderRegistry constructs an empty registry.

func (*MetricsRecorderRegistry) Lookup added in v0.7.0 

func (r *MetricsRecorderRegistry) Lookup(name string) (MetricsRecorder, bool)

Lookup retrieves a recorder by name.

func (*MetricsRecorderRegistry) Register added in v0.7.0 

func (r *MetricsRecorderRegistry) Register(name string, mr MetricsRecorder) error

Register stores a recorder by name.

func (*MetricsRecorderRegistry) RegisterNamespaced added in v0.7.0 

func (r *MetricsRecorderRegistry) RegisterNamespaced(namespace, name string, mr MetricsRecorder) error

RegisterNamespaced stores a recorder by namespace+name.

func (*MetricsRecorderRegistry) SetNamespacer added in v0.7.0 

func (r *MetricsRecorderRegistry) SetNamespacer(fn func(string, string) string)

SetNamespacer customizes namespacing.

type MuxResolver added in v0.7.0 

type MuxResolver[T any] struct {
	// contains filtered or unexported fields
}

MuxResolver resolves handlers from a router mux using the message type.

func NewMuxResolver added in v0.7.0 

func NewMuxResolver[T any](mux *router.Mux) *MuxResolver[T]

NewMuxResolver builds a resolver using the provided mux (or a new one when nil).

func (*MuxResolver[T]) Resolve added in v0.7.0 

func (r *MuxResolver[T]) Resolve(_ context.Context, msg T) ([]command.Commander[T], error)

Resolve looks up handlers by message type and converts them to Commanders.

type NodeLayout added in v0.15.0 

type NodeLayout struct {
	X      float64 `json:"x"`
	Y      float64 `json:"y"`
	Width  float64 `json:"width"`
	Height float64 `json:"height"`
	ZIndex int     `json:"z_index,omitempty"`
}

type NonRetryableError added in v0.15.0 

type NonRetryableError struct {
	Message string
	Cause   error
}

NonRetryableError marks terminal errors for retry classifiers.

func (*NonRetryableError) Error added in v0.15.0 

func (e *NonRetryableError) Error() string

func (*NonRetryableError) Unwrap added in v0.15.0 

func (e *NonRetryableError) Unwrap() error

type Option added in v0.7.0 

type Option = runner.Option

Option mirrors runner.Option so flows can share runner configuration knobs.

type Orchestrator added in v0.15.0 

type Orchestrator[T any] interface {
	Start(ctx context.Context, req StartRequest[T]) (*ExecutionHandle, error)
	Pause(ctx context.Context, executionID string) error
	Resume(ctx context.Context, executionID string) error
	Stop(ctx context.Context, executionID string) error
	Status(ctx context.Context, executionID string) (*ExecutionStatus, error)
}

Orchestrator executes transition effects according to policy depth.

type OutboxDispatcher added in v0.15.0 

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

OutboxDispatcher publishes pending outbox entries to a scheduler.

func NewOutboxDispatcher added in v0.15.0 

func NewOutboxDispatcher(
	store OutboxStore,
	scheduler JobScheduler,
	opts ...OutboxDispatcherOption,
) *OutboxDispatcher

NewOutboxDispatcher constructs a dispatcher for durable outbox processing.

func (*OutboxDispatcher) DispatchPending added in v0.15.0 

func (d *OutboxDispatcher) DispatchPending(ctx context.Context) (int, error)

DispatchPending claims pending entries and enqueues them into the scheduler.

func (*OutboxDispatcher) Health added in v0.15.0 

func (d *OutboxDispatcher) Health(ctx context.Context) DispatcherHealth

Health returns a derived health summary and emits health hooks.

func (*OutboxDispatcher) Run added in v0.15.0 

func (d *OutboxDispatcher) Run(ctx context.Context) error

Run starts continuous dispatch polling until context cancellation or Stop.

func (*OutboxDispatcher) RunOnce added in v0.15.0 

func (d *OutboxDispatcher) RunOnce(ctx context.Context) (DispatchReport, error)

RunOnce executes one claim/enqueue/ack cycle and classifies outcomes.

func (*OutboxDispatcher) Status added in v0.15.0 

func (d *OutboxDispatcher) Status() DispatcherRuntimeStatus

Status returns a copy of the latest runtime status.

func (*OutboxDispatcher) Stop added in v0.15.0 

func (d *OutboxDispatcher) Stop(ctx context.Context) error

Stop requests background loop termination and waits for graceful stop.

type OutboxDispatcherOption added in v0.15.0 

type OutboxDispatcherOption func(*OutboxDispatcher)

OutboxDispatcherOption customizes dispatcher behavior.

func WithOutboxBackoff added in v0.15.0 

func WithOutboxBackoff(fn func(attempt int, baseDelay time.Duration) time.Duration) OutboxDispatcherOption

WithOutboxBackoff customizes retry schedule per attempt.

func WithOutboxHealthHook added in v0.15.0 

func WithOutboxHealthHook(hook func(context.Context, DispatcherHealth)) OutboxDispatcherOption

WithOutboxHealthHook receives health snapshots after each cycle.

func WithOutboxLeaseDuration added in v0.15.0 

func WithOutboxLeaseDuration(dur time.Duration) OutboxDispatcherOption

WithOutboxLeaseDuration sets lease expiration for claimed entries.

func WithOutboxLimit added in v0.15.0 

func WithOutboxLimit(limit int) OutboxDispatcherOption

WithOutboxLimit sets the max entries claimed per dispatch call.

func WithOutboxLogger added in v0.15.0 

func WithOutboxLogger(logger Logger) OutboxDispatcherOption

WithOutboxLogger configures dispatcher logging.

func WithOutboxMaxAttempts added in v0.15.0 

func WithOutboxMaxAttempts(maxAttempts int) OutboxDispatcherOption

WithOutboxMaxAttempts sets the terminal-attempt threshold before dead-lettering.

func WithOutboxMetrics added in v0.15.0 

func WithOutboxMetrics(metrics DispatcherMetrics) OutboxDispatcherOption

WithOutboxMetrics configures dispatcher metrics recording hooks.

func WithOutboxOutcomeHook added in v0.15.0 

func WithOutboxOutcomeHook(hook func(context.Context, DispatchEntryResult)) OutboxDispatcherOption

WithOutboxOutcomeHook receives one callback per classified dispatch outcome.

func WithOutboxRetryDelay added in v0.15.0 

func WithOutboxRetryDelay(delay time.Duration) OutboxDispatcherOption

WithOutboxRetryDelay sets retry schedule for failed dispatches.

func WithOutboxRetryOwner added in v0.15.0 

func WithOutboxRetryOwner(owner DispatchRetryOwner) OutboxDispatcherOption

WithOutboxRetryOwner configures single-owner retry/backoff authority.

func WithOutboxRunInterval added in v0.15.0 

func WithOutboxRunInterval(interval time.Duration) OutboxDispatcherOption

WithOutboxRunInterval sets background runner poll cadence.

func WithOutboxStatusHook added in v0.15.0 

func WithOutboxStatusHook(hook func(context.Context, DispatcherRuntimeStatus)) OutboxDispatcherOption

WithOutboxStatusHook receives runtime status updates.

func WithOutboxWorkerID added in v0.15.0 

func WithOutboxWorkerID(workerID string) OutboxDispatcherOption

WithOutboxWorkerID overrides the worker identifier used by ClaimPending.

type OutboxEntry added in v0.15.0 

type OutboxEntry struct {
	ID           string
	EntityID     string
	TransitionID string
	ExecutionID  string
	Event        string
	Topic        string
	Payload      []byte
	Effect       Effect
	Status       string
	Attempts     int
	LeaseOwner   string
	LeaseUntil   time.Time
	LeaseToken   string
	RetryAt      time.Time
	CreatedAt    time.Time
	ProcessedAt  *time.Time
	LastError    string
	Metadata     map[string]any
}

OutboxEntry stores durable effect descriptors emitted with state commits.

type OutboxStore added in v0.15.0 

type OutboxStore interface {
	ClaimPending(ctx context.Context, workerID string, limit int, leaseTTL time.Duration) ([]ClaimedOutboxEntry, error)
	MarkCompleted(ctx context.Context, id, leaseToken string) error
	MarkFailed(ctx context.Context, id, leaseToken string, retryAt time.Time, reason string) error
	MarkDeadLetter(ctx context.Context, id, leaseToken, reason string) error
	ListDeadLetters(ctx context.Context, scope DeadLetterScope) ([]OutboxEntry, error)
	ExtendLease(ctx context.Context, id, leaseToken string, leaseTTL time.Duration) error
}

OutboxStore exposes lease/claim/retry operations for dispatch loops.

type ParallelConfig added in v0.7.0 

type ParallelConfig struct {
	Steps         []string    `json:"steps" yaml:"steps"`
	ErrorStrategy string      `json:"error_strategy,omitempty" yaml:"error_strategy,omitempty"`
	Opts          FlowOptions `json:"options,omitempty" yaml:"options,omitempty"`
}

type ParallelExecutor 

type ParallelExecutor[T any] struct {
	// contains filtered or unexported fields
}

func NewParallelExecutor 

func NewParallelExecutor[T any](handlers []command.Commander[T], opts ...runner.Option) *ParallelExecutor[T]

NewParallelExecutor creates a new ParallelExecutor with the provided handlers

func (*ParallelExecutor[T]) Execute 

func (p *ParallelExecutor[T]) Execute(ctx context.Context, msg T) error

func (*ParallelExecutor[T]) WithErrorStrategy added in v0.7.0 

func (p *ParallelExecutor[T]) WithErrorStrategy(strategy ErrorStrategy) *ParallelExecutor[T]

WithErrorStrategy overrides the error strategy used to combine handler errors.

type Point added in v0.15.0 

type Point struct {
	X float64 `json:"x"`
	Y float64 `json:"y"`
}

type RPCErrorEnvelope added in v0.15.0 

type RPCErrorEnvelope struct {
	Code      string         `json:"code"`
	Message   string         `json:"message"`
	Category  string         `json:"category,omitempty"`
	Retryable bool           `json:"retryable,omitempty"`
	Details   map[string]any `json:"details,omitempty"`
}

RPCErrorEnvelope is the internal RPC transport error shape.

func RPCErrorForError added in v0.15.0 

func RPCErrorForError(err error) *RPCErrorEnvelope

RPCErrorForError returns a canonical RPC envelope for engine/runtime errors.

type RedisClient added in v0.7.0 

type RedisClient interface {
	EvalSHA(ctx context.Context, sha string, keys []string, args ...any) (any, error)
	Eval(ctx context.Context, script string, keys []string, args ...any) (any, error)
	ScriptLoad(ctx context.Context, script string) (string, error)
}

RedisClient captures script primitives required for distributed-safe state/outbox operations.

type RedisStateStore added in v0.7.0 

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

RedisStateStore persists state and outbox records via Redis Lua scripts.

func NewRedisStateStore added in v0.7.0 

func NewRedisStateStore(client RedisClient, ttl time.Duration) *RedisStateStore

NewRedisStateStore builds a store using the provided script-capable redis client.

func (*RedisStateStore) ClaimPending added in v0.15.0 

func (s *RedisStateStore) ClaimPending(
	ctx context.Context,
	workerID string,
	limit int,
	leaseTTL time.Duration,
) ([]ClaimedOutboxEntry, error)

ClaimPending claims pending entries with leased ownership tokens.

func (*RedisStateStore) ExtendLease added in v0.15.0 

func (s *RedisStateStore) ExtendLease(ctx context.Context, id, leaseToken string, leaseTTL time.Duration) error

ExtendLease extends one claimed entry lease.

func (*RedisStateStore) ListDeadLetters added in v0.15.0 

func (s *RedisStateStore) ListDeadLetters(ctx context.Context, scope DeadLetterScope) ([]OutboxEntry, error)

ListDeadLetters returns dead-lettered entries from redis-backed outbox storage.

func (*RedisStateStore) Load added in v0.7.0 

func (s *RedisStateStore) Load(ctx context.Context, id string) (*StateRecord, error)

Load reads one state record from redis.

func (*RedisStateStore) MarkCompleted added in v0.15.0 

func (s *RedisStateStore) MarkCompleted(ctx context.Context, id, leaseToken string) error

MarkCompleted marks one claimed entry as completed after verifying lease token ownership.

func (*RedisStateStore) MarkDeadLetter added in v0.15.0 

func (s *RedisStateStore) MarkDeadLetter(ctx context.Context, id, leaseToken, reason string) error

MarkDeadLetter marks one claimed entry as dead-lettered after verifying lease ownership.

func (*RedisStateStore) MarkFailed added in v0.15.0 

func (s *RedisStateStore) MarkFailed(ctx context.Context, id, leaseToken string, retryAt time.Time, reason string) error

MarkFailed marks one claimed entry for retry after verifying lease token ownership.

func (*RedisStateStore) RunInTransaction added in v0.15.0 

func (s *RedisStateStore) RunInTransaction(ctx context.Context, fn func(TxStore) error) error

RunInTransaction stages one state write + N outbox writes and commits atomically.

func (*RedisStateStore) SaveIfVersion added in v0.15.0 

func (s *RedisStateStore) SaveIfVersion(ctx context.Context, rec *StateRecord, expectedVersion int) (int, error)

SaveIfVersion performs optimistic compare-and-set using one Lua script.

type Resolver added in v0.7.0 

type Resolver[T any] interface {
	Resolve(ctx context.Context, msg T) ([]command.Commander[T], error)
}

Resolver returns the handlers that should run for a given message.

type ResolverMap added in v0.15.0 

type ResolverMap[T any] struct {
	// contains filtered or unexported fields
}

ResolverMap is a simple in-memory resolver registry implementation.

func NewResolverMap added in v0.15.0 

func NewResolverMap[T any]() *ResolverMap[T]

NewResolverMap creates an empty resolver registry.

func (*ResolverMap[T]) DynamicTarget added in v0.15.0 

func (r *ResolverMap[T]) DynamicTarget(ref string) (DynamicTargetResolver[T], bool)

DynamicTarget resolves a dynamic target resolver by reference.

func (*ResolverMap[T]) DynamicTargetIDs added in v0.15.0 

func (r *ResolverMap[T]) DynamicTargetIDs() []string

DynamicTargetIDs returns sorted dynamic target resolver identifiers.

func (*ResolverMap[T]) Guard added in v0.15.0 

func (r *ResolverMap[T]) Guard(ref string) (Guard[T], bool)

Guard resolves a guard by reference.

func (*ResolverMap[T]) GuardIDs added in v0.15.0 

func (r *ResolverMap[T]) GuardIDs() []string

GuardIDs returns sorted guard resolver identifiers.

func (*ResolverMap[T]) RegisterDynamicTarget added in v0.15.0 

func (r *ResolverMap[T]) RegisterDynamicTarget(ref string, resolver DynamicTargetResolver[T])

RegisterDynamicTarget stores a dynamic target resolver.

func (*ResolverMap[T]) RegisterGuard added in v0.15.0 

func (r *ResolverMap[T]) RegisterGuard(ref string, guard Guard[T])

RegisterGuard stores a guard resolver.

type ResolverRegistry added in v0.15.0 

type ResolverRegistry[T any] interface {
	Guard(ref string) (Guard[T], bool)
	DynamicTarget(ref string) (DynamicTargetResolver[T], bool)
}

ResolverRegistry resolves runtime guard and dynamic target references.

type ResumeRequest added in v0.15.0 

type ResumeRequest[T command.Message] struct {
	ExecutionID     string
	EntityID        string
	Event           string
	Msg             T
	ExpectedState   string
	ExpectedVersion int
	ExecCtx         ExecutionContext
	Apply           func(context.Context, ApplyEventRequest[T]) (*ApplyEventResponse[T], error)
}

ResumeRequest encapsulates a resume callback with stale-state safeguards.

type RetryPolicy added in v0.15.0 

type RetryPolicy struct {
	WorkerRetries bool
	TaskRetries   bool
}

RetryPolicy captures retry ownership settings for one execution path.

func (RetryPolicy) Validate added in v0.15.0 

func (p RetryPolicy) Validate() error

Validate enforces the single retry owner rule.

type RetryableFlow added in v0.7.0 

type RetryableFlow[T command.Message] struct {
	// contains filtered or unexported fields
}

Example usage of retry with backoff for any flow pattern

func NewRetryableFlow added in v0.7.0 

func NewRetryableFlow[T command.Message](
	flow Flow[T],
	retryStrategy runner.RetryStrategy,
	maxRetries int,
) *RetryableFlow[T]

func (*RetryableFlow[T]) Execute added in v0.7.0 

func (r *RetryableFlow[T]) Execute(ctx context.Context, msg T) error

type SQLiteStateStore added in v0.7.0 

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

SQLiteStateStore persists records and outbox entries in SQLite.

func NewSQLiteStateStore added in v0.7.0 

func NewSQLiteStateStore(db *sql.DB, table string) *SQLiteStateStore

NewSQLiteStateStore builds a store using the given DB and state table name.

func (*SQLiteStateStore) ClaimPending added in v0.15.0 

func (s *SQLiteStateStore) ClaimPending(
	ctx context.Context,
	workerID string,
	limit int,
	leaseTTL time.Duration,
) ([]ClaimedOutboxEntry, error)

ClaimPending claims pending outbox entries with a worker lease.

func (*SQLiteStateStore) ExtendLease added in v0.15.0 

func (s *SQLiteStateStore) ExtendLease(ctx context.Context, id, leaseToken string, leaseTTL time.Duration) error

ExtendLease extends lease ownership for one claimed outbox entry.

func (*SQLiteStateStore) ListDeadLetters added in v0.15.0 

func (s *SQLiteStateStore) ListDeadLetters(ctx context.Context, scope DeadLetterScope) ([]OutboxEntry, error)

ListDeadLetters returns dead-lettered outbox entries for inspection.

func (*SQLiteStateStore) Load added in v0.7.0 

func (s *SQLiteStateStore) Load(ctx context.Context, id string) (*StateRecord, error)

Load reads state for entity id.

func (*SQLiteStateStore) MarkCompleted added in v0.15.0 

func (s *SQLiteStateStore) MarkCompleted(ctx context.Context, id, leaseToken string) error

MarkCompleted marks one outbox entry as completed.

func (*SQLiteStateStore) MarkDeadLetter added in v0.15.0 

func (s *SQLiteStateStore) MarkDeadLetter(ctx context.Context, id, leaseToken, reason string) error

MarkDeadLetter marks one outbox entry as terminal after verifying lease ownership.

func (*SQLiteStateStore) MarkFailed added in v0.15.0 

func (s *SQLiteStateStore) MarkFailed(ctx context.Context, id, leaseToken string, retryAt time.Time, reason string) error

MarkFailed marks one outbox entry as pending retry.

func (*SQLiteStateStore) RunInTransaction added in v0.15.0 

func (s *SQLiteStateStore) RunInTransaction(ctx context.Context, fn func(TxStore) error) error

RunInTransaction executes fn in a DB transaction.

func (*SQLiteStateStore) SaveIfVersion added in v0.15.0 

func (s *SQLiteStateStore) SaveIfVersion(ctx context.Context, rec *StateRecord, expectedVersion int) (int, error)

SaveIfVersion writes record using optimistic version compare.

type Saga added in v0.7.0 

type Saga[T command.Message] struct {
	// contains filtered or unexported fields
}

Saga implements the saga pattern for distributed transactions a compensating transaction must be idempotent and retryable

func NewSaga added in v0.7.0 

func NewSaga[T command.Message](steps []SagaStep[T], compensate bool) *Saga[T]

func (*Saga[T]) Execute added in v0.7.0 

func (s *Saga[T]) Execute(ctx context.Context, msg T) error

type SagaConfig added in v0.7.0 

type SagaConfig struct {
	Steps             []SagaStepConfig `json:"steps" yaml:"steps"`
	CompensateOnError bool             `json:"compensate_on_error,omitempty" yaml:"compensate_on_error,omitempty"`
}

type SagaStep added in v0.7.0 

type SagaStep[T command.Message] struct {
	Name       string
	Execute    func(context.Context, T) error
	Compensate func(context.Context, T) error
}

type SagaStepConfig added in v0.7.0 

type SagaStepConfig struct {
	Do         string `json:"do" yaml:"do"`
	Compensate string `json:"compensate,omitempty" yaml:"compensate,omitempty"`
}

type ScheduledExecutionMessage added in v0.15.0 

type ScheduledExecutionMessage struct {
	Message   *ExecutionMessage
	ExecuteAt time.Time
}

ScheduledExecutionMessage captures scheduled jobs for in-memory tests.

type SerialConfig added in v0.7.0 

type SerialConfig struct {
	Steps []string    `json:"steps" yaml:"steps"`
	Opts  FlowOptions `json:"options,omitempty" yaml:"options,omitempty"`
}

type SerialExecutor added in v0.7.0 

type SerialExecutor[T any] struct {
	// contains filtered or unexported fields
}

SerialExecutor runs multiple command handlers in sequence

func NewSerialExecutor added in v0.7.0 

func NewSerialExecutor[T any](handlers []command.Commander[T], opts ...runner.Option) *SerialExecutor[T]

NewSerialExecutor creates a new SerialExecutor with the provided handlers

func (*SerialExecutor[T]) Execute added in v0.7.0 

func (s *SerialExecutor[T]) Execute(ctx context.Context, msg T) error

type Snapshot added in v0.15.0 

type Snapshot struct {
	EntityID           string
	CurrentState       string
	AllowedTransitions []TransitionInfo
	Metadata           map[string]any
}

Snapshot captures current state and transition metadata.

type SnapshotRequest added in v0.15.0 

type SnapshotRequest[T any] struct {
	MachineID      string
	EntityID       string
	Msg            T
	ExecCtx        ExecutionContext
	EvaluateGuards bool
	IncludeBlocked bool
}

SnapshotRequest is the canonical request envelope for snapshot reads.

type StartRequest added in v0.15.0 

type StartRequest[T any] struct {
	ExecutionID     string
	MachineID       string
	MachineVersion  string
	EntityID        string
	Event           string
	TransitionID    string
	PreviousState   string
	CurrentState    string
	ExpectedState   string
	ExpectedVersion int
	ExecCtx         ExecutionContext
	RetryPolicy     RetryPolicy
	Result          *TransitionResult[T]
	Snapshot        *Snapshot
	Msg             T
	Metadata        map[string]any
}

StartRequest carries transition output into the orchestration layer.

type State added in v0.15.0 

type State struct {
	Name     string
	Initial  bool
	Metadata map[string]any
}

State is a compiled runtime state.

type StateConfig added in v0.7.0 

type StateConfig struct {
	Name        string `json:"name" yaml:"name"`
	Description string `json:"description,omitempty" yaml:"description,omitempty"`
	Terminal    bool   `json:"terminal,omitempty" yaml:"terminal,omitempty"`
	Initial     bool   `json:"initial,omitempty" yaml:"initial,omitempty"`
}

type StateDefinition added in v0.15.0 

type StateDefinition struct {
	Name     string
	Initial  bool
	Terminal bool
	Metadata map[string]any
}

StateDefinition is a canonical authoring state.

type StateMachine added in v0.7.0 

type StateMachine[T command.Message] struct {
	// contains filtered or unexported fields
}

StateMachine executes transitions using compiled contracts and versioned persistence.

func NewStateMachineFromDefinition added in v0.15.0 

func NewStateMachineFromDefinition[T command.Message](
	def *MachineDefinition,
	store StateStore,
	req TransitionRequest[T],
	resolvers ResolverRegistry[T],
	actions *ActionRegistry[T],
	opts ...StateMachineOption[T],
) (*StateMachine[T], error)

NewStateMachineFromDefinition constructs a state machine from canonical definition.

func (*StateMachine[T]) ApplyEvent added in v0.15.0 

func (s *StateMachine[T]) ApplyEvent(ctx context.Context, req ApplyEventRequest[T]) (*ApplyEventResponse[T], error)

ApplyEvent applies the provided event to current state and returns transition envelope.

func (*StateMachine[T]) Execute added in v0.7.0 

func (s *StateMachine[T]) Execute(ctx context.Context, msg T) error

Execute is the compatibility wrapper for command.Commander[T].

func (*StateMachine[T]) ExecutionHistory added in v0.15.0 

func (s *StateMachine[T]) ExecutionHistory(ctx context.Context, scope ExecutionScope) ([]TransitionLifecycleEvent[T], error)

ExecutionHistory returns lifecycle events matching the provided scope.

func (*StateMachine[T]) ExecutionList added in v0.15.0 

func (s *StateMachine[T]) ExecutionList(ctx context.Context, scope ExecutionScope) ([]ExecutionStatus, error)

ExecutionList returns execution statuses matching the provided scope.

func (*StateMachine[T]) ExecutionStatus added in v0.15.0 

func (s *StateMachine[T]) ExecutionStatus(ctx context.Context, executionID string) (*ExecutionStatus, error)

ExecutionStatus returns current orchestration status for an execution.

func (*StateMachine[T]) PauseExecution added in v0.15.0 

func (s *StateMachine[T]) PauseExecution(ctx context.Context, executionID string) error

PauseExecution pauses one orchestrated execution by identifier.

func (*StateMachine[T]) ResumeExecution added in v0.15.0 

func (s *StateMachine[T]) ResumeExecution(ctx context.Context, executionID string) error

ResumeExecution resumes one orchestrated execution by identifier.

func (*StateMachine[T]) Snapshot added in v0.15.0 

func (s *StateMachine[T]) Snapshot(ctx context.Context, req SnapshotRequest[T]) (*Snapshot, error)

Snapshot returns transition metadata for the current entity state.

func (*StateMachine[T]) StopExecution added in v0.15.0 

func (s *StateMachine[T]) StopExecution(ctx context.Context, executionID string) error

StopExecution stops one orchestrated execution by identifier.

type StateMachineConfig added in v0.7.0 

type StateMachineConfig struct {
	Entity          string             `json:"entity" yaml:"entity"`
	ExecutionPolicy ExecutionPolicy    `json:"execution_policy" yaml:"execution_policy"`
	HookFailureMode HookFailureMode    `json:"hook_failure_mode,omitempty" yaml:"hook_failure_mode,omitempty"`
	States          []StateConfig      `json:"states" yaml:"states"`
	Transitions     []TransitionConfig `json:"transitions" yaml:"transitions"`
	PersistWith     string             `json:"persist_with,omitempty" yaml:"persist_with,omitempty"`
}

func (StateMachineConfig) ToMachineDefinition added in v0.15.0 

func (s StateMachineConfig) ToMachineDefinition() *MachineDefinition

ToMachineDefinition emits a canonical machine definition for runtime assembly.

func (StateMachineConfig) Validate added in v0.7.0 

func (s StateMachineConfig) Validate() error

Validate ensures the state machine definition is well formed.

type StateMachineOption added in v0.7.0 

type StateMachineOption[T command.Message] func(*StateMachine[T])

StateMachineOption customizes state machine behavior.

func WithExecutionPolicy added in v0.15.0 

func WithExecutionPolicy[T command.Message](policy ExecutionPolicy) StateMachineOption[T]

WithExecutionPolicy selects the runtime orchestration policy.

func WithHookFailureMode added in v0.15.0 

func WithHookFailureMode[T command.Message](mode HookFailureMode) StateMachineOption[T]

WithHookFailureMode configures lifecycle hook error behavior.

func WithIdempotencyStore added in v0.15.0 

func WithIdempotencyStore[T command.Message](store IdempotencyStore[T]) StateMachineOption[T]

WithIdempotencyStore sets an explicit idempotency store implementation.

func WithLifecycleHooks added in v0.15.0 

func WithLifecycleHooks[T command.Message](hooks ...TransitionLifecycleHook[T]) StateMachineOption[T]

WithLifecycleHooks configures transition lifecycle hooks.

func WithLogger added in v0.15.0 

func WithLogger[T command.Message](logger Logger) StateMachineOption[T]

WithLogger sets the state-machine logger.

func WithOrchestrator added in v0.15.0 

func WithOrchestrator[T command.Message](orchestrator Orchestrator[T]) StateMachineOption[T]

WithOrchestrator sets an explicit orchestrator implementation.

type StateNodeSchema added in v0.15.0 

type StateNodeSchema struct {
	ID       string      `json:"id"`
	Label    string      `json:"label"`
	Terminal bool        `json:"terminal"`
	Initial  bool        `json:"initial,omitempty"`
	UI       UIComponent `json:"ui"`
}

type StateRecord added in v0.15.0 

type StateRecord struct {
	EntityID       string
	State          string
	Version        int
	MachineID      string
	MachineVersion string
	Metadata       map[string]any
	UpdatedAt      time.Time
}

StateRecord is the persisted state row for an entity.

type StateStore added in v0.7.0 

type StateStore interface {
	Load(ctx context.Context, id string) (*StateRecord, error)
	SaveIfVersion(ctx context.Context, rec *StateRecord, expectedVersion int) (newVersion int, err error)
	RunInTransaction(ctx context.Context, fn func(TxStore) error) error
}

StateStore persists state records with optimistic locking and transactional outbox writes.

type Step added in v0.15.0 

type Step struct {
	ActionID string
	Async    bool
	Delay    time.Duration
	Timeout  time.Duration
	Metadata map[string]any
}

Step is a compiled workflow step descriptor.

type StepDefinition added in v0.15.0 

type StepDefinition struct {
	ActionID string
	Async    bool
	Delay    string
	Timeout  string
	Metadata map[string]any
}

StepDefinition is a canonical authoring workflow step.

type StepUISchema added in v0.15.0 

type StepUISchema struct {
	Type       string         `json:"type"`
	Properties map[string]any `json:"properties,omitempty"`
	UI         UIComponent    `json:"ui"`
}

type TargetInfo added in v0.15.0 

type TargetInfo struct {
	Kind       string
	To         string
	Resolver   string
	Resolved   bool
	ResolvedTo string
	Candidates []string
}

TargetInfo captures static/dynamic target metadata.

type TargetUISchema added in v0.15.0 

type TargetUISchema struct {
	Kind       string   `json:"kind"`
	To         string   `json:"to,omitempty"`
	Resolver   string   `json:"resolver,omitempty"`
	Candidates []string `json:"candidates,omitempty"`
}

type TransitionConfig added in v0.7.0 

type TransitionConfig struct {
	Name   string `json:"name" yaml:"name"`
	From   string `json:"from" yaml:"from"`
	To     string `json:"to" yaml:"to"`
	Guard  string `json:"guard,omitempty" yaml:"guard,omitempty"`
	Action string `json:"action,omitempty" yaml:"action,omitempty"`
}

type TransitionDefinition added in v0.15.0 

type TransitionDefinition struct {
	ID        string
	Event     string
	From      string
	To        string
	DynamicTo *DynamicTargetDefinition
	Guards    []GuardDefinition
	Workflow  TransitionWorkflowDefinition
	Metadata  map[string]any
}

TransitionDefinition is a canonical authoring transition.

type TransitionInfo added in v0.15.0 

type TransitionInfo struct {
	ID         string
	Event      string
	Target     TargetInfo
	Allowed    bool
	Rejections []GuardRejection
	Metadata   map[string]any
}

TransitionInfo describes one transition available from snapshot state.

type TransitionLifecycleEvent added in v0.15.0 

type TransitionLifecycleEvent[T any] struct {
	Phase           TransitionPhase
	MachineID       string
	MachineVersion  string
	EntityID        string
	ExecutionID     string
	Event           string
	TransitionID    string
	PreviousState   string
	CurrentState    string
	ExpectedState   string
	ExpectedVersion int
	ErrorCode       string
	ErrorMessage    string
	ExecCtx         ExecutionContext
	Metadata        map[string]any
	OccurredAt      time.Time
	Msg             T
}

TransitionLifecycleEvent captures auditable transition metadata.

type TransitionLifecycleHook added in v0.15.0 

type TransitionLifecycleHook[T any] interface {
	Notify(ctx context.Context, evt TransitionLifecycleEvent[T]) error
}

TransitionLifecycleHook receives transition lifecycle events.

type TransitionLifecycleHooks added in v0.15.0 

type TransitionLifecycleHooks[T any] []TransitionLifecycleHook[T]

TransitionLifecycleHooks fan-out collection for lifecycle hooks.

type TransitionPhase added in v0.15.0 

type TransitionPhase string

TransitionPhase identifies lifecycle event emission points. 

const (
	TransitionPhaseAttempted TransitionPhase = "attempted"
	TransitionPhaseCommitted TransitionPhase = "committed"
	TransitionPhaseRejected  TransitionPhase = "rejected"
)

type TransitionRequest added in v0.7.0 

type TransitionRequest[T any] struct {
	StateKey func(T) string
	Event    func(T) string
}

TransitionRequest extracts state machine metadata from a message.

type TransitionResult added in v0.15.0 

type TransitionResult[T any] struct {
	PreviousState string
	CurrentState  string
	Effects       []Effect
}

TransitionResult captures transition execution outcome.

type TransitionSchema added in v0.15.0 

type TransitionSchema struct {
	ID       string           `json:"id"`
	Event    string           `json:"event"`
	From     string           `json:"from"`
	Target   TargetUISchema   `json:"target"`
	Guards   []GuardUISchema  `json:"guards,omitempty"`
	Workflow WorkflowUISchema `json:"workflow"`
	Metadata map[string]any   `json:"metadata,omitempty"`
}

type TransitionWorkflowDefinition added in v0.15.0 

type TransitionWorkflowDefinition struct {
	Nodes []WorkflowNodeDefinition
}

TransitionWorkflowDefinition is a declarative transition workflow.

type TransportErrorMapping added in v0.15.0 

type TransportErrorMapping struct {
	RuntimeCode string
	HTTPStatus  int
	GRPCCode    string
	RPCCode     string
}

TransportErrorMapping defines protocol-level mappings for runtime errors.

func MapRuntimeError added in v0.15.0 

func MapRuntimeError(err error) TransportErrorMapping

MapRuntimeError maps canonical runtime categories to transport protocol categories.

type TxStore added in v0.15.0 

type TxStore interface {
	Load(ctx context.Context, id string) (*StateRecord, error)
	SaveIfVersion(ctx context.Context, rec *StateRecord, expectedVersion int) (newVersion int, err error)
	AppendOutbox(ctx context.Context, entry OutboxEntry) error
}

TxStore is the transactional state store boundary.

type UIComponent added in v0.15.0 

type UIComponent struct {
	Component string         `json:"component"`
	Layout    string         `json:"layout,omitempty"`
	Config    map[string]any `json:"config,omitempty"`
}

type ValidationDiagnostic added in v0.15.0 

type ValidationDiagnostic struct {
	Code     string `json:"code"`
	Severity string `json:"severity"`
	Message  string `json:"message"`
	Path     string `json:"path"`
	NodeID   string `json:"node_id,omitempty"`
	Field    string `json:"field,omitempty"`
}

ValidationDiagnostic is a deterministic validation message for editor/runtime tooling.

func ValidateMachineDefinition added in v0.15.0 

func ValidateMachineDefinition(def *MachineDefinition, catalog *EditorCatalog) []ValidationDiagnostic

ValidateMachineDefinition validates canonical definitions deterministically.

func ValidateMachineDefinitionScoped added in v0.15.0 

func ValidateMachineDefinitionScoped(def *MachineDefinition, catalog *EditorCatalog, scope *ValidationScope) []ValidationDiagnostic

ValidateMachineDefinitionScoped validates full machine or changed-node scope only.

type ValidationScope added in v0.15.0 

type ValidationScope struct {
	NodeIDs []string
}

ValidationScope limits validation emissions to changed editor nodes.

type Viewport added in v0.15.0 

type Viewport struct {
	X    float64 `json:"x"`
	Y    float64 `json:"y"`
	Zoom float64 `json:"zoom"`
}

type WorkflowNodeDefinition added in v0.15.0 

type WorkflowNodeDefinition struct {
	ID       string
	Kind     string
	Step     *StepDefinition
	Expr     string
	Next     []string
	Metadata map[string]any
}

WorkflowNodeDefinition describes a declarative workflow graph node.

type WorkflowNodeUISchema added in v0.15.0 

type WorkflowNodeUISchema struct {
	ID        string        `json:"id"`
	Kind      string        `json:"kind"`
	Step      *StepUISchema `json:"step,omitempty"`
	Condition string        `json:"condition,omitempty"`
	Next      []string      `json:"next,omitempty"`
	UI        UIComponent   `json:"ui"`
}

type WorkflowUISchema added in v0.15.0 

type WorkflowUISchema struct {
	Nodes []WorkflowNodeUISchema `json:"nodes"`
}

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.